一、Planner在Agent系统中的位置

前六篇已经完成了Agent基础循环、工具注册、参数校验、工具集成、A2A与MCP等能力。

接下来需要解决一个更核心的问题：

面对一个复杂目标，Agent如何知道应该先做什么、后做什么？

例如用户提出：

分析华东区域本月销售下降原因，并生成一份改进报告。

这不是一次工具调用能完成的任务。

合理步骤可能包括：

查询销售数据
→ 查询库存与退货
→ 对比历史同期
→ 找出异常区域和产品
→ 形成原因假设
→ 生成改进建议
→ 输出报告

Planner负责把模糊目标转换为可执行计划。

二、生产级Planner需要满足什么要求

一个可用Planner至少应具备：

1. 输出结构化计划；
2. 每个步骤有唯一ID；
3. 明确步骤依赖；
4. 不允许循环依赖；
5. 只能引用已注册工具；
6. 能标记风险等级；
7. 每一步有可验证输出；
8. 限制最大步骤数；
9. 计划失败可以修复；
10. 可以持久化和恢复。

调用链：

用户目标
→ Planner生成JSON
→ Schema校验
→ DAG校验
→ 工具与风险校验
→ 保存计划
→ Executor执行

三、项目结构

agent-planner
├── requirements.txt
├── .env.example
└── src
    ├── models.py
    ├── llm_client.py
    ├── parser.py
    ├── validator.py
    ├── planner.py
    └── main.py

依赖：

pydantic>=2.12,<3
httpx>=0.28,<1
python-dotenv>=1.2,<2

四、定义计划数据模型

models.py：

from __future__ import annotations

from enum import StrEnum

from pydantic import BaseModel, Field


class RiskLevel(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class PlanStep(BaseModel):
    id: str = Field(
        pattern=r"^S[1-9][0-9]*$",
        description="步骤ID，例如S1、S2"
    )
    title: str = Field(min_length=1, max_length=80)
    description: str = Field(min_length=1, max_length=500)
    dependencies: list[str] = Field(default_factory=list)
    tool_hint: str | None = Field(
        default=None,
        description="建议工具，必须来自工具白名单"
    )
    expected_output: str = Field(
        min_length=1,
        max_length=300
    )
    risk_level: RiskLevel = RiskLevel.LOW
    max_retries: int = Field(default=1, ge=0, le=3)


class TaskPlan(BaseModel):
    objective: str = Field(min_length=1, max_length=500)
    assumptions: list[str] = Field(default_factory=list)
    steps: list[PlanStep] = Field(min_length=1, max_length=20)
    completion_criteria: list[str] = Field(min_length=1)

为什么不用自由字典？

因为执行器需要稳定协议：

Planner
→ Executor
→ Reviewer
→ Persistence

任何字段漂移都会影响后续模块。

五、定义大模型客户端

为了不绑定具体厂商，先定义协议。

llm_client.py：

from __future__ import annotations

from typing import Protocol

import httpx


class LlmClient(Protocol):
    def complete(
        self,
        system_prompt: str,
        user_prompt: str
    ) -> str:
        ...


class OpenAiCompatibleClient:
    def __init__(
        self,
        base_url: str,
        api_key: str,
        model: str,
        timeout_seconds: float = 90.0
    ) -> None:
        self._base_url = base_url.rstrip("/")
        self._api_key = api_key
        self._model = model
        self._client = httpx.Client(
            timeout=httpx.Timeout(timeout_seconds)
        )

    def complete(
        self,
        system_prompt: str,
        user_prompt: str
    ) -> str:
        response = self._client.post(
            f"{self._base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self._api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": self._model,
                "temperature": 0.1,
                "messages": [
                    {
                        "role": "system",
                        "content": system_prompt
                    },
                    {
                        "role": "user",
                        "content": user_prompt
                    }
                ]
            }
        )

        response.raise_for_status()
        payload = response.json()

        return payload["choices"][0]["message"]["content"]

业务层只依赖 LlmClient，后续可以替换：

• DeepSeek；
• OpenAI；
• Qwen；
• 本地vLLM；
• 测试Mock。

六、编写Planner提示词

planner.py中的System Prompt：

PLANNER_SYSTEM_PROMPT = """
你是生产级AI Agent任务规划器。

你的职责是把用户目标拆解为可执行、可验证的步骤。

规则：
1. 只输出一个JSON对象。
2. 不要输出Markdown代码块、解释、前言或结尾。
3. steps数量必须为1到20。
4. 步骤ID必须为S1、S2、S3格式。
5. dependencies只能引用计划中的步骤ID。
6. 不允许循环依赖。
7. 每一步必须有明确expected_output。
8. tool_hint只能从允许工具列表中选择，也可以为null。
9. 读取操作通常为low；修改操作至少为medium；
   支付、删除、权限变更必须为high。
10. 不得假设已经获得用户未提供的敏感信息。
11. 高风险动作只负责规划，执行阶段必须人工审批。
"""

用户Prompt：

def build_plan_prompt(
    objective: str,
    allowed_tools: set[str]
) -> str:
    tools_text = ", ".join(sorted(allowed_tools))

    return f"""
用户目标：
{objective}

允许使用的工具：
{tools_text}

输出结构：
{{
  "objective": "最终目标",
  "assumptions": ["必要假设"],
  "steps": [
    {{
      "id": "S1",
      "title": "步骤标题",
      "description": "具体执行说明",
      "dependencies": [],
      "tool_hint": null,
      "expected_output": "可验证输出",
      "risk_level": "low",
      "max_retries": 1
    }}
  ],
  "completion_criteria": ["完成标准"]
}}
"""

Temperature设置为0.1，降低格式漂移。

七、实现JSON提取器

parser.py：

from __future__ import annotations

import json
import re

from pydantic import ValidationError

from models import TaskPlan


class PlanParseError(ValueError):
    pass


def strip_markdown_fence(text: str) -> str:
    value = text.strip()

    match = re.fullmatch(
        r"```(?:json)?\s*(.*?)\s*```",
        value,
        flags=re.DOTALL | re.IGNORECASE
    )

    if match:
        return match.group(1).strip()

    return value


def extract_first_json_object(text: str) -> str:
    start = text.find("{")

    if start < 0:
        raise PlanParseError("模型响应中没有JSON对象")

    depth = 0
    in_string = False
    escape = False

    for index in range(start, len(text)):
        char = text[index]

        if escape:
            escape = False
            continue

        if in_string and char == "\\":
            escape = True
            continue

        if char == '"':
            in_string = not in_string
            continue

        if in_string:
            continue

        if char == "{":
            depth += 1
        elif char == "}":
            depth -= 1

            if depth == 0:
                return text[start:index + 1]

    raise PlanParseError("JSON对象没有正确闭合")


def parse_task_plan(raw_text: str) -> TaskPlan:
    clean = strip_markdown_fence(raw_text)
    json_text = extract_first_json_object(clean)

    try:
        payload = json.loads(json_text)
    except json.JSONDecodeError as exc:
        raise PlanParseError(
            f"JSON解析失败：第{exc.lineno}行"
            f"第{exc.colno}列，{exc.msg}"
        ) from exc

    try:
        return TaskPlan.model_validate(payload)
    except ValidationError as exc:
        raise PlanParseError(
            f"Schema校验失败：{exc.errors()}"
        ) from exc

不要使用 eval() 解析模型输出。

八、实现计划校验器

validator.py：

from __future__ import annotations

from collections import defaultdict, deque

from models import RiskLevel, TaskPlan


class PlanValidationError(ValueError):
    pass


class PlanValidator:
    def __init__(
        self,
        allowed_tools: set[str]
    ) -> None:
        self._allowed_tools = allowed_tools

    def validate(self, plan: TaskPlan) -> None:
        self._validate_unique_ids(plan)
        self._validate_dependencies(plan)
        self._validate_acyclic(plan)
        self._validate_tools(plan)
        self._validate_risk(plan)

    @staticmethod
    def _validate_unique_ids(plan: TaskPlan) -> None:
        ids = [step.id for step in plan.steps]

        if len(ids) != len(set(ids)):
            raise PlanValidationError("存在重复步骤ID")

    @staticmethod
    def _validate_dependencies(plan: TaskPlan) -> None:
        known = {step.id for step in plan.steps}

        for step in plan.steps:
            if step.id in step.dependencies:
                raise PlanValidationError(
                    f"{step.id}不能依赖自己"
                )

            unknown = set(step.dependencies) - known

            if unknown:
                raise PlanValidationError(
                    f"{step.id}引用未知依赖："
                    f"{sorted(unknown)}"
                )

    @staticmethod
    def _validate_acyclic(plan: TaskPlan) -> None:
        graph: dict[str, list[str]] = defaultdict(list)
        indegree = {
            step.id: 0 for step in plan.steps
        }

        for step in plan.steps:
            for dependency in step.dependencies:
                graph[dependency].append(step.id)
                indegree[step.id] += 1

        queue = deque(
            step_id
            for step_id, degree in indegree.items()
            if degree == 0
        )

        visited = 0

        while queue:
            current = queue.popleft()
            visited += 1

            for next_step in graph[current]:
                indegree[next_step] -= 1

                if indegree[next_step] == 0:
                    queue.append(next_step)

        if visited != len(plan.steps):
            raise PlanValidationError(
                "计划中存在循环依赖"
            )

    def _validate_tools(self, plan: TaskPlan) -> None:
        for step in plan.steps:
            if (
                step.tool_hint
                and step.tool_hint not in self._allowed_tools
            ):
                raise PlanValidationError(
                    f"{step.id}使用未注册工具："
                    f"{step.tool_hint}"
                )

    @staticmethod
    def _validate_risk(plan: TaskPlan) -> None:
        high_risk_keywords = {
            "支付",
            "删除",
            "退款",
            "修改权限",
            "发送外部邮件",
            "执行SQL写入"
        }

        for step in plan.steps:
            content = (
                step.title + " " + step.description
            )

            if (
                any(word in content for word in high_risk_keywords)
                and step.risk_level != RiskLevel.HIGH
            ):
                raise PlanValidationError(
                    f"{step.id}包含高风险动作，"
                    "但risk_level不是high"
                )

九、实现自动修复

模型第一次输出失败时，把错误反馈给模型。

def build_repair_prompt(
    objective: str,
    invalid_output: str,
    error_message: str,
    allowed_tools: set[str]
) -> str:
    return f"""
上一次任务计划没有通过系统校验。

原始目标：
{objective}

允许工具：
{", ".join(sorted(allowed_tools))}

校验错误：
{error_message}

无效输出：
{invalid_output}

请修复所有错误。
只输出完整JSON对象。
"""

完整Planner：

from __future__ import annotations

from llm_client import LlmClient
from models import TaskPlan
from parser import PlanParseError, parse_task_plan
from validator import PlanValidationError, PlanValidator


class TaskPlanner:
    def __init__(
        self,
        client: LlmClient,
        allowed_tools: set[str],
        max_repair_attempts: int = 1
    ) -> None:
        self._client = client
        self._allowed_tools = allowed_tools
        self._validator = PlanValidator(allowed_tools)
        self._max_repair_attempts = max_repair_attempts

    def create_plan(self, objective: str) -> TaskPlan:
        raw = self._client.complete(
            PLANNER_SYSTEM_PROMPT,
            build_plan_prompt(
                objective,
                self._allowed_tools
            )
        )

        for attempt in range(
            self._max_repair_attempts + 1
        ):
            try:
                plan = parse_task_plan(raw)
                self._validator.validate(plan)
                return plan
            except (
                PlanParseError,
                PlanValidationError
            ) as exc:
                if attempt >= self._max_repair_attempts:
                    raise RuntimeError(
                        f"无法生成有效计划：{exc}"
                    ) from exc

                raw = self._client.complete(
                    PLANNER_SYSTEM_PROMPT,
                    build_repair_prompt(
                        objective=objective,
                        invalid_output=raw,
                        error_message=str(exc),
                        allowed_tools=self._allowed_tools
                    )
                )

        raise RuntimeError("Planner进入不可达状态")

修复次数必须有限，防止无限消耗。

十、增加拓扑排序

Executor需要知道可执行顺序。

def topological_order(plan: TaskPlan) -> list[str]:
    graph: dict[str, list[str]] = defaultdict(list)
    indegree = {
        step.id: 0 for step in plan.steps
    }

    for step in plan.steps:
        for dependency in step.dependencies:
            graph[dependency].append(step.id)
            indegree[step.id] += 1

    queue = deque(
        step_id
        for step_id, degree in indegree.items()
        if degree == 0
    )

    result: list[str] = []

    while queue:
        current = queue.popleft()
        result.append(current)

        for next_step in graph[current]:
            indegree[next_step] -= 1

            if indegree[next_step] == 0:
                queue.append(next_step)

    if len(result) != len(plan.steps):
        raise PlanValidationError("无法拓扑排序")

    return result

无依赖步骤可以并行，但生产系统还要考虑：

• 同一资源写冲突；
• API限流；
• 用户预算；
• 工具并发安全；
• 事务顺序。

十一、主程序

main.py：

from __future__ import annotations

import os

from dotenv import load_dotenv

from llm_client import OpenAiCompatibleClient
from planner import TaskPlanner
from validator import topological_order


def main() -> None:
    load_dotenv()

    api_key = os.environ.get("LLM_API_KEY")

    if not api_key:
        raise RuntimeError(
            "缺少环境变量LLM_API_KEY"
        )

    client = OpenAiCompatibleClient(
        base_url=os.environ.get(
            "LLM_BASE_URL",
            "https://api.deepseek.com"
        ),
        api_key=api_key,
        model=os.environ.get(
            "LLM_MODEL",
            "deepseek-v4-flash"
        )
    )

    allowed_tools = {
        "search_knowledge",
        "query_sales",
        "query_inventory",
        "query_returns",
        "generate_report"
    }

    planner = TaskPlanner(
        client=client,
        allowed_tools=allowed_tools,
        max_repair_attempts=1
    )

    objective = (
        "分析华东区域本月销售下降原因，"
        "并生成一份改进报告"
    )

    plan = planner.create_plan(objective)

    print(plan.model_dump_json(indent=2))
    print("执行顺序：", topological_order(plan))


if __name__ == "__main__":
    main()

.env.example：

LLM_BASE_URL=https://api.deepseek.com
LLM_API_KEY=
LLM_MODEL=deepseek-v4-flash

十二、可能的输出

{
  "objective": "分析华东区域本月销售下降原因并生成改进报告",
  "assumptions": [
    "系统中存在销售、库存和退货数据"
  ],
  "steps": [
    {
      "id": "S1",
      "title": "查询销售数据",
      "description": "获取华东区域本月及历史同期销售数据",
      "dependencies": [],
      "tool_hint": "query_sales",
      "expected_output": "按区域和产品汇总的销售数据",
      "risk_level": "low",
      "max_retries": 1
    },
    {
      "id": "S2",
      "title": "查询库存数据",
      "description": "获取华东区域库存与缺货记录",
      "dependencies": [],
      "tool_hint": "query_inventory",
      "expected_output": "库存和缺货汇总",
      "risk_level": "low",
      "max_retries": 1
    },
    {
      "id": "S3",
      "title": "查询退货数据",
      "description": "分析退货率和主要退货原因",
      "dependencies": [],
      "tool_hint": "query_returns",
      "expected_output": "退货数据与原因分布",
      "risk_level": "low",
      "max_retries": 1
    },
    {
      "id": "S4",
      "title": "分析下降原因",
      "description": "综合销售、库存和退货数据定位异常",
      "dependencies": ["S1", "S2", "S3"],
      "tool_hint": null,
      "expected_output": "按影响程度排序的原因列表",
      "risk_level": "low",
      "max_retries": 1
    },
    {
      "id": "S5",
      "title": "生成改进报告",
      "description": "形成原因、证据、建议和跟踪指标",
      "dependencies": ["S4"],
      "tool_hint": "generate_report",
      "expected_output": "完整经营改进报告",
      "risk_level": "low",
      "max_retries": 1
    }
  ],
  "completion_criteria": [
    "所有原因都有数据证据",
    "每项建议都有责任对象和指标",
    "报告生成成功"
  ]
}

执行顺序：

S1、S2、S3
→ S4
→ S5

前三步可以并行。

十三、计划如何持久化

建议状态表：

agent_plan
├── plan_id
├── trace_id
├── conversation_id
├── objective
├── plan_json
├── plan_version
├── status
├── created_at
└── updated_at

步骤表：

agent_plan_step
├── plan_id
├── step_id
├── status
├── input_json
├── output_json
├── retry_count
├── started_at
├── completed_at
└── error_message

状态：

PENDING
READY
RUNNING
WAITING_APPROVAL
SUCCEEDED
FAILED
SKIPPED

恢复时不要重新生成全部计划，先读取持久化状态。

十四、什么时候需要重规划

以下情况可以触发Replan：

• 工具不可用；
• 关键假设不成立；
• 用户修改目标；
• 数据不足；
• 某步骤永久失败；
• 发现新的必要步骤。

Replanner输入应包括：

原目标
当前计划
已完成步骤
失败步骤
工具结果
剩余预算

重规划后必须：

• 保留已完成结果；
• 增加plan_version；
• 记录变更原因；
• 重新进行全部校验；
• 不允许重复高风险操作。

十五、Planner的安全边界

Planner只能提出计划，不能直接授权。

例如：

{
  "title": "删除异常订单",
  "risk_level": "high"
}

执行器必须进入：

权限校验
→ 参数范围校验
→ 人工审批
→ 幂等校验
→ 工具执行

不要因为Planner标记了“high”就认为风险已经处理。

十六、生产环境还需要补齐什么

本文实现了Planner内核，后续还需增加：

• Planner自动化测试集；
• 计划质量评分；
• 任务预算；
• Executor；
• Reviewer；
• Replanner；
• 并发调度；
• 人工审批；
• Trace与回放；
• 数据库持久化；
• 多租户隔离。

下一篇将继续实现：

Planner、Executor与Reviewer三层架构，让任务计划真正进入执行闭环。

总结

生产级Planner不是一句Prompt，而是一条受控管道：

目标
→ 模型生成
→ JSON提取
→ Schema校验
→ DAG校验
→ 工具白名单
→ 风险规则
→ 有限修复
→ 持久化
→ 执行

模型负责提供智能，程序负责保证计划可执行、可验证和安全。

参考资料

• https://arxiv.org/abs/2210.03629
• https://blog.langchain.com/planning-agents/
• https://docs.pydantic.dev/latest/