优先级排序（Prioritization）| Agentic Design Patterns | 智能体设计模式

作者： /
内容纲要

一、核心定义

优先级排序：智能体根据重要性、紧迫性、依赖关系等标准评估和排序任务/目标,动态确定下一步行动,确保资源用在刀刃上。

核心问题：

无优先级排序  →  盲目执行任务,效率低,目标不达
有优先级排序  →  聚焦关键任务,资源优化,目标对齐

工作机制：

多个潜在任务
  ↓
定义评估标准(紧迫性/重要性/依赖/成本)
  ↓
对每个任务评分
  ↓
排序并选择最优
  ↓
动态调整(环境变化时)
  ↓
执行高优先级任务

二、为什么需要优先级排序？

痛点 影响 解决方案
任务过多 不知从何做起 按优先级排序
目标冲突 无法兼顾所有 选择最重要的
资源受限 时间/计算/人力不足 优先分配关键任务
延迟/失败 关键任务被耽误 紧急任务优先
效率低下 做了不重要的事 聚焦高价值任务

核心优势：

  • ✅ 提高执行效率
  • ✅ 确保目标对齐
  • ✅ 优化资源利用
  • ✅ 增强适应性
  • ✅ 降低失败风险

三、优先级排序的四大要素

1. 标准定义（Criteria Definition）

常用标准：

标准 说明 示例
紧迫性（Urgency） 时间敏感度 系统故障 > 功能优化
重要性（Importance） 对目标的影响 核心功能 > 辅助功能
依赖关系（Dependencies） 是否是其他任务前提 数据库设计 > 业务逻辑
资源可用性 所需工具/信息是否就绪 有API Key的任务优先
成本/收益（Cost/Benefit） 投入与产出比 ROI高的任务优先
用户偏好 个性化优先级 VIP客户请求优先

标准组合：

# 示例：综合评分公式
priority_score = (
    urgency_score * 0.4 +      # 紧迫性权重40%
    importance_score * 0.3 +   # 重要性权重30%
    dependency_score * 0.2 +   # 依赖性权重20%
    resource_score * 0.1       # 资源可用性权重10%
)

2. 任务评估（Task Evaluation）

评估方法：

方法 实现 适用场景
基于规则 if-else逻辑 简单、明确的标准
评分算法 加权计算 多标准综合评估
LLM推理 用大模型分析 复杂、模糊的情况
机器学习 训练分类模型 有历史数据

LLM评估示例：

PRIORITY_ANALYSIS_PROMPT = """
请评估以下任务的优先级：

任务：{task_description}
当前情况：{context}

评估维度：
1. 紧迫性 (1-5分)
2. 重要性 (1-5分)
3. 依赖关系
4. 资源需求

输出格式：
- 紧迫性：[分数] 理由：...
- 重要性：[分数] 理由：...
- 建议优先级：P0/P1/P2
"""

3. 调度/选择逻辑（Scheduling Logic）

常见算法：

算法 原理 适用场景
FIFO（先进先出） 按到达顺序 公平处理所有任务
优先级队列 高优先级先执行 有明确优先级
轮询（Round Robin） 轮流处理 防止饥饿
最短作业优先 快速任务先做 提高吞吐量
截止期限优先 临近deadline先做 时间约束强

示例：优先级队列：

import heapq

class PriorityTaskQueue:
    def __init__(self):
        self.tasks = []

    def add_task(self, priority, task):
        # priority越小,优先级越高
        heapq.heappush(self.tasks, (priority, task))

    def get_next_task(self):
        if self.tasks:
            return heapq.heappop(self.tasks)[1]
        return None

# 使用
queue = PriorityTaskQueue()
queue.add_task(priority=0, task="紧急故障修复")  # P0
queue.add_task(priority=1, task="功能开发")      # P1
queue.add_task(priority=2, task="文档更新")      # P2

next_task = queue.get_next_task()  # 返回"紧急故障修复"

4. 动态重新排序（Dynamic Re-prioritization）

触发条件：

  • 新的紧急任务出现
  • 截止日期临近
  • 资源状态变化
  • 依赖任务完成/失败
  • 环境条件改变

示例：

class AdaptiveAgent:
    def __init__(self):
        self.task_queue = PriorityTaskQueue()

    def re_evaluate_priorities(self, event):
        """根据事件重新评估优先级"""
        if event.type == "新紧急任务":
            # 插入最高优先级
            self.task_queue.add_task(priority=0, task=event.task)

        elif event.type == "截止期限临近":
            # 提升相关任务优先级
            self.boost_task_priority(event.task_id)

        elif event.type == "资源可用":
            # 重新计算所有任务优先级
            self.recalculate_all_priorities()

四、优先级排序的三个层次

图1：优先级排序设计模式

1. 高层目标优先级（Strategic Level）

场景：多个战略目标
示例：
  目标A：提升用户留存(重要性5,紧迫性3)
  目标B：降低运营成本(重要性4,紧迫性2)
  目标C：修复安全漏洞(重要性5,紧迫性5)

优先级排序：C > A > B

2. 子任务优先级（Tactical Level）

场景：执行计划中的步骤排序
示例（安全漏洞修复）：
  步骤1：识别漏洞(依赖前置)
  步骤2：设计修复方案(依赖步骤1)
  步骤3：实施修复(依赖步骤2)
  步骤4：测试验证(依赖步骤3)

顺序执行，严格依赖

3. 行动选择（Operational Level）

场景：即时决策
示例：
  选项A：查询数据库(资源可用,1秒)
  选项B：调用外部API(需要等待,5秒)
  选项C：使用缓存(立即,0.1秒)

优先：C > A > B

五、典型应用场景

1. 自动化客户支持

场景：多个客户工单
优先级规则：
  P0：系统故障报告(紧急)
  P1：高价值客户请求(重要)
  P2：常规密码重置(一般)

效果：确保关键问题及时处理

2. 云计算资源调度

场景：多应用竞争资源
优先级规则：
  高峰期：关键应用优先
  低峰期：批处理任务优先

效果：优化成本,保证核心业务

3. 自动驾驶系统

场景：多个并发决策
优先级：
  P0：避免碰撞(安全)
  P1：保持车道(稳定)
  P2：燃油优化(效率)

原则：安全 > 稳定 > 效率

4. 金融交易

场景：多个交易机会
评估维度：
  - 市场波动性
  - 风险承受度
  - 利润率
  - 实时新闻

动态调整：根据市场变化实时排序

5. 项目管理

场景：任务看板
评估标准：
  - 截止日期
  - 依赖关系
  - 团队可用性
  - 战略重要性

工具：Kanban Board自动排序

6. 网络安全

场景：多个安全警报
优先级：
  P0：高威胁+关键资产
  P1：中威胁+重要资产
  P2：低威胁

目标：资源用于最危险威胁

7. 个人助理AI

场景：日程管理
优先级：
  - 用户定义重要性
  - 截止日期
  - 当前情境(位置/时间)

示例：会议 > 提醒 > 通知

六、实战案例：项目经理智能体

系统设计

任务数据模型：

from pydantic import BaseModel

class Task(BaseModel):
    id: str                    # 任务ID
    description: str           # 描述
    priority: Optional[str]    # P0/P1/P2
    assigned_to: Optional[str] # 分配对象

任务管理器：

class SuperSimpleTaskManager:
    def __init__(self):
        self.tasks: Dict[str, Task] = {}
        self.next_task_id = 1

    def create_task(self, description: str) -> Task:
        """创建新任务"""
        task_id = f"TASK-{self.next_task_id:03d}"
        new_task = Task(id=task_id, description=description)
        self.tasks[task_id] = new_task
        self.next_task_id += 1
        return new_task

    def update_task(self, task_id: str, **kwargs) -> Task:
        """更新任务"""
        task = self.tasks.get(task_id)
        if task:
            updated_task = task.model_copy(update=kwargs)
            self.tasks[task_id] = updated_task
            return updated_task
        return None

    def list_all_tasks(self) -> str:
        """列出所有任务"""
        return "\n".join([
            f"ID: {t.id}, Desc: '{t.description}', "
            f"Priority: {t.priority or 'N/A'}, "
            f"Assigned To: {t.assigned_to or 'N/A'}"
            for t in self.tasks.values()
        ])

智能体工具

from langchain_core.tools import Tool

pm_tools = [
    Tool(
        name="create_new_task",
        func=lambda desc: task_manager.create_task(desc),
        description="创建新任务并获取ID"
    ),
    Tool(
        name="assign_priority_to_task",
        func=lambda task_id, priority: 
            task_manager.update_task(task_id, priority=priority),
        description="分配优先级（P0/P1/P2）"
    ),
    Tool(
        name="assign_task_to_worker",
        func=lambda task_id, worker: 
            task_manager.update_task(task_id, assigned_to=worker),
        description="分配任务给工作人员"
    ),
    Tool(
        name="list_all_tasks",
        func=task_manager.list_all_tasks,
        description="列出所有任务"
    ),
]

智能体Prompt

PM_PROMPT = """
你是项目经理AI智能体。任务：高效管理项目任务。

收到新任务请求时,按以下步骤：
1. 首先用 create_new_task 创建任务并获取 task_id
2. 分析请求:
   - 如果提到"urgent/ASAP/critical" → 设为P0
   - 如果提到工作人员 → 使用 assign_task_to_worker
3. 如果信息缺失:
   - 默认优先级：P1
   - 默认分配：Worker A
4. 完成后用 list_all_tasks 显示最终状态

可用人员：Worker A, Worker B, Review Team
优先级：P0(最高), P1(中), P2(最低)
"""

运行示例

# 场景1：紧急任务
输入："实现新登录系统,很急,分配给Worker B"

Agent思考过程：
  1. 创建任务 → TASK-001
  2. 识别关键词"很急" → 设为P0
  3. 识别"Worker B" → 分配给Worker B
  4. 输出任务列表

输出：
  TASK-001: 实现新登录系统
  Priority: P0
  Assigned To: Worker B

# 场景2：一般任务
输入："审查营销网站内容"

Agent思考过程：
  1. 创建任务 → TASK-002
  2. 无紧急关键词 → 默认P1
  3. 无指定人员 → 默认Worker A
  4. 输出任务列表

输出：
  TASK-002: 审查营销网站内容
  Priority: P1
  Assigned To: Worker A

七、优势与挑战

✅ 核心优势

优势 说明
提升效率 优先做重要的事
目标对齐 确保战略执行
资源优化 最大化ROI
增强适应性 动态调整策略
降低风险 关键任务不遗漏

⚠️ 挑战

挑战 影响 缓解方案
标准定义困难 难以量化重要性 结合专家知识+LLM
动态环境复杂 频繁重排序消耗资源 设定重排序阈值
饥饿问题 低优先级任务永不执行 老化机制(等待越久优先级越高)
冲突解决 同优先级如何选择 引入次级排序标准
评估成本 评估本身消耗时间 缓存评分结果

八、常见问题

Q1：如何防止低优先级任务饥饿？
实施"老化"机制：任务等待时间越长,优先级自动提升。例如每等待1小时,优先级+1分。

Q2：同优先级任务如何排序？
引入次级标准：1) 创建时间(FIFO)；2) 执行时长(短任务优先)；3) 依赖关系；4) 随机选择。

Q3：如何平衡紧迫性和重要性？
使用二维矩阵(艾森豪威尔矩阵)：

  • 重要且紧急 → P0
  • 重要不紧急 → P1
  • 紧急不重要 → P2
  • 不重要不紧急 → P3

Q4：动态重排序的频率？
事件驱动：只在特定事件触发时重排序,避免持续计算开销。

Q5：如何评估优先级系统效果？
指标：1) 关键任务完成率；2) 平均响应时间；3) 资源利用率；4) 目标达成度。

Q6：LLM评估vs规则评估？

  • 规则：快速、可预测、适合明确场景
  • LLM：灵活、理解复杂语义、成本较高
  • 推荐：简单任务用规则,复杂任务用LLM

九、参考资源

学术论文

原文章节

  • <code>26-Chapter-20-Prioritization.md</code>

相关章节

  • 第 6 章：规划 - 优先级排序在规划中的应用
  • 第 7 章：多智能体协作 - 团队任务分配
  • 第 16 章：资源感知优化 - 资源约束下的优化
  • 第 17 章：推理技术 - 推理辅助优先级决策

相关文章

滚动至顶部