# 第 20 章：优先级排序

在复杂、动态的环境中，智能体常常面临大量潜在行动、相互冲突的目标以及有限的资源。如果没有明确的流程来确定后续行动，智能体可能会遇到效率低下、操作延迟或无法实现关键目标的问题。优先级排序模式通过使智能体根据重要性、紧迫性、依赖关系和既定标准来评估和排序任务、目标或行动，从而解决这一挑战。这确保智能体将精力集中在最关键的任务上，从而提高有效性和目标一致性。

## 优先级排序模式概述

智能体使用优先级排序来有效管理任务、目标和子目标，指导后续行动。这个过程有助于在处理多个需求时做出明智决策，优先处理重要或紧急的活动，而不是次要的活动。这在资源受限、时间有限以及目标可能冲突的实际场景中尤为重要。

智能体优先级排序通常包含几个核心要素。首先是**标准定义**，即建立任务评估的规则或指标，这些标准可能包括：紧急性（任务的时间敏感性）、重要性（对主要目标的影响）、依赖关系（该任务是否是其他任务的前提）、资源可用性（必要工具或信息的就绪状态）、成本/收益分析（投入与预期成果的对比），以及面向个性化智能体的用户偏好。其次是**任务评估**，即根据上述标准评估每个潜在任务，评估方法可以从简单规则到大语言模型的复杂评分或推理。第三是**调度或选择逻辑**，即基于评估结果选择最佳下一步行动或任务序列的算法，可能使用队列或高级规划组件。最后是**动态重新优先级排序**，允许智能体根据情况变化（如新关键事件出现或截止日期临近）调整优先级，确保智能体的适应性和响应能力。

优先级排序可以在各个层面进行：选择总体目标（高层次目标优先级排序）、在计划内排序步骤（子任务优先级排序）或从可用选项中选择下一个即时行动（行动选择）。有效的优先级排序使智能体展现更智能、更高效、更稳健的行为，特别是在复杂的多目标环境中。这反映了人类团队的组织方式，其中管理者通过考虑所有成员的输入来优先处理任务。

## 实际应用和用例

在各种实际应用中，AI 智能体展现了优先级排序的复杂运用，以做出及时有效的决策。

* **自动化客户支持**：智能体优先处理紧急请求，如系统停机报告，而不是日常事务，如密码重置。他们还可能优先处理高价值客户。
* **云计算**：AI 通过优先将资源分配给高峰需求期间的关键应用程序来管理和调度资源，同时将不太紧急的批处理作业推迟到非高峰时段以优化成本。
* **自动驾驶系统**：持续优先处理行动以确保安全和效率。例如，为避免碰撞而刹车优先于保持车道纪律或优化燃油效率。
* **金融交易**：交易机器人通过分析市场条件、风险承受能力、利润率和实时新闻等因素来优先处理交易，实现高优先级交易的快速执行。
* **项目管理**：AI 智能体根据截止日期、依赖关系、团队可用性和战略重要性来优先处理项目板上的任务。
* **网络安全**：监控网络流量的智能体根据威胁严重性、潜在影响和资产关键性来优先处理警报，确保对最危险威胁的即时响应。
* **个人助理 AI**：利用优先级排序来管理日常生活，根据用户定义的重要性、即将到来的截止日期和当前上下文来组织日历事件、提醒和通知。

这些示例共同说明了优先级排序能力对于增强 AI 智能体在各种情况下的性能和决策能力是多么基础。

## 实践代码示例

以下演示了使用 LangChain 开发项目管理 AI 智能体。该智能体负责任务的创建、优先级排序和分配给团队成员，说明了大语言模型与定制工具在自动化项目管理中的应用。

```python
import os
import asyncio
from typing import List, Optional, Dict, Type
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain.memory import ConversationBufferMemory

## --- 0. 配置和设置 ---
## 从 .env 文件加载 OPENAI_API_KEY。
load_dotenv()

## ChatOpenAI 客户端自动从环境中获取 API 密钥。
llm = ChatOpenAI(temperature=0.5, model="gpt-4o-mini")

## --- 1. 任务管理系统 ---
class Task(BaseModel):
    """表示系统中的单个任务。"""
    id: str
    description: str
    priority: Optional[str] = None  # P0, P1, P2
    assigned_to: Optional[str] = None # 工作人员的名字

class SuperSimpleTaskManager:
    """一个高效且稳健的内存任务管理器。"""
    def __init__(self):
        # 使用字典实现 O(1) 查找、更新和删除。
        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
        print(f"DEBUG: 任务已创建 - {task_id}: {description}")
        return new_task

    def update_task(self, task_id: str, **kwargs) -> Optional[Task]:
        """使用 Pydantic 的 model_copy 安全地更新任务。"""
        task = self.tasks.get(task_id)
        if task:
            # 使用 model_copy 进行类型安全的更新。
            update_data = {k: v for k, v in kwargs.items() if v is not None}
            updated_task = task.model_copy(update=update_data)
            self.tasks[task_id] = updated_task
            print(f"DEBUG: 任务 {task_id} 已更新为 {update_data}")
            return updated_task
        print(f"DEBUG: 未找到任务 {task_id} 进行更新。")
        return None

    def list_all_tasks(self) -> str:
        """列出系统中当前的所有任务。"""
        if not self.tasks:
            return "系统中没有任务。"

        task_strings = []
        for task in self.tasks.values():
            task_strings.append(
                f"ID: {task.id}, 描述: '{task.description}', "
                f"优先级: {task.priority or 'N/A'}, "
                f"分配给: {task.assigned_to or 'N/A'}"
            )
        return "当前任务:\n" + "\n".join(task_strings)

task_manager = SuperSimpleTaskManager()

## --- 2. 项目管理智能体工具 ---
## 使用 Pydantic 模型作为工具参数以获得更好的验证和清晰度。
class CreateTaskArgs(BaseModel):
    description: str = Field(description="任务的详细描述。")

class PriorityArgs(BaseModel):
    task_id: str = Field(description="要更新的任务 ID，例如 'TASK-001'。")
    priority: str = Field(description="要设置的优先级。必须是以下之一：'P0'、'P1'、'P2'。")

class AssignWorkerArgs(BaseModel):
    task_id: str = Field(description="要更新的任务 ID，例如 'TASK-001'。")
    worker_name: str = Field(description="要分配任务的工作人员的名字。")

def create_new_task_tool(description: str) -> str:
    """使用给定的描述创建一个新的项目任务。"""
    task = task_manager.create_task(description)
    return f"已创建任务 {task.id}: '{task.description}'。"

def assign_priority_to_task_tool(task_id: str, priority: str) -> str:
    """为给定的任务 ID 分配优先级（P0、P1、P2）。"""
    if priority not in ["P0", "P1", "P2"]:
        return "优先级无效。必须是 P0、P1 或 P2。"
    task = task_manager.update_task(task_id, priority=priority)
    return f"已为任务 {task.id} 分配优先级 {priority}。" if task else f"未找到任务 {task_id}。"

def assign_task_to_worker_tool(task_id: str, worker_name: str) -> str:
    """将任务分配给特定的工作人员。"""
    task = task_manager.update_task(task_id, assigned_to=worker_name)
    return f"已将任务 {task.id} 分配给 {worker_name}。" if task else f"未找到任务 {task_id}。"

## PM 智能体使用的所有工具
pm_tools = [
    Tool(
        name="create_new_task",
        func=create_new_task_tool,
        description="首先使用此工具创建一个新任务并获取其 ID。",
        args_schema=CreateTaskArgs
    ),
    Tool(
        name="assign_priority_to_task",
        func=assign_priority_to_task_tool,
        description="使用此工具在创建任务后为其分配优先级。",
        args_schema=PriorityArgs
    ),
    Tool(
        name="assign_task_to_worker",
        func=assign_task_to_worker_tool,
        description="使用此工具在创建任务后将其分配给特定的工作人员。",
        args_schema=AssignWorkerArgs
    ),
    Tool(
        name="list_all_tasks",
        func=task_manager.list_all_tasks,
        description="使用此工具列出所有当前任务及其状态。"
    ),
]

## --- 3. 项目管理智能体定义 ---
pm_prompt_template = ChatPromptTemplate.from_messages([
    ("system", """你是一个专注的项目管理 LLM 智能体。你的目标是高效地管理项目任务。
       当你收到新的任务请求时，遵循以下步骤：
    1.  首先，使用 `create_new_task` 工具创建具有给定描述的任务。你必须首先执行此操作以获取 `task_id`。
    2.  接下来，分析用户的请求以查看是否提到了优先级或受让人。
        - 如果提到优先级（例如，"紧急"、"ASAP"、"关键"），将其映射到 P0。使用 `assign_priority_to_task`。
        - 如果提到工作人员，使用 `assign_task_to_worker`。
    3.  如果缺少任何信息（优先级、受让人），你必须做出合理的默认分配（例如，分配 P1 优先级并分配给 'Worker A'）。
    4.  一旦任务完全处理完毕，使用 `list_all_tasks` 显示最终状态。
       可用的工作人员：'Worker A'、'Worker B'、'Review Team'
    优先级级别：P0（最高）、P1（中等）、P2（最低）
    """),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

## 创建智能体执行器
pm_agent = create_react_agent(llm, pm_tools, pm_prompt_template)
pm_agent_executor = AgentExecutor(
    agent=pm_agent,
    tools=pm_tools,
    verbose=True,
    handle_parsing_errors=True,
    memory=ConversationBufferMemory(memory_key="chat_history", return_messages=True)
)

## --- 4. 简单交互流程 ---
async def run_simulation():
    print("--- 项目管理模拟 ---")
    # 场景 1：处理新的紧急功能请求
    print("\n[用户请求] 我需要尽快实现一个新的登录系统。它应该分配给 Worker B。")
    await pm_agent_executor.ainvoke({"input": "创建一个实现新登录系统的任务。这很紧急，应该分配给 Worker B。"})
    print("\n" + "-"*60 + "\n")

    # 场景 2：处理细节较少的不太紧急的内容更新
    print("[用户请求] 我们需要审查营销网站内容。")
    await pm_agent_executor.ainvoke({"input": "管理一个新任务：审查营销网站内容。"})
    print("\n--- 模拟完成 ---")

## 运行模拟
if __name__ == "__main__":
    asyncio.run(run_simulation())
```

此代码使用 Python 和 LangChain 实现了一个简单的任务管理系统，旨在模拟由大语言模型驱动的项目管理智能体。

该系统采用 [`SuperSimpleTaskManager`](chapters/Chapter 20_ Prioritization.md:58) 类在内存中高效管理任务，利用字典结构实现快速数据检索。每个任务由 [`Task`](chapters/Chapter 20_ Prioritization.md:51) Pydantic 模型表示，包含唯一标识符、描述文本、可选优先级级别（P0、P1、P2）和可选的受让人指定等属性。内存使用量会根据任务类型、工作人员数量等因素而变化。任务管理器提供任务创建、任务修改和检索所有任务的方法。

智能体通过一组定义好的工具与任务管理器交互。这些工具支持创建新任务、为任务分配优先级、将任务分配给人员以及列出所有任务。每个工具都被封装为与 [`SuperSimpleTaskManager`](chapters/Chapter 20_ Prioritization.md:58) 的实例交互。Pydantic 模型用于描述工具所需的参数，从而确保数据验证。

[`AgentExecutor`](chapters/Chapter 20_ Prioritization.md:179) 配置了语言模型、工具集和对话内存组件以保持上下文连续性。定义了特定的 [`ChatPromptTemplate`](chapters/Chapter 20_ Prioritization.md:161) 来指导智能体在项目管理角色中的行为。提示词指示智能体先创建任务，然后根据指定分配优先级和人员，最后以完整的任务列表结束。当信息缺失时，提示词中规定了默认分配，例如 P1 优先级和 'Worker A'。

代码包含一个异步模拟函数（[`run_simulation`](chapters/Chapter 20_ Prioritization.md:189)）来演示智能体的协作能力。模拟执行两个不同的场景：管理带有指定人员的紧急任务，以及管理输入信息较少的非紧急任务。由于在 [`AgentExecutor`](chapters/Chapter 20_ Prioritization.md:179) 中启用了 `verbose=True`，智能体的行动和逻辑过程会输出到控制台。

## 概览

**定义（What）：** 在复杂环境中运行的 AI 智能体面临大量潜在行动、相互冲突的目标和有限的资源。如果没有明确的方法来确定下一步行动，这些智能体可能会变得低效且效果不佳，甚至导致严重的操作延迟或完全无法完成主要目标。核心挑战是管理这一数量庞大的选择，确保智能体有目的地、合乎逻辑地行动。

**原因（Why）：** 优先级排序模式通过使智能体对任务和目标进行排序，为这个问题提供了标准化的解决方案。这是通过建立明确的标准（如紧急性、重要性、依赖关系和资源成本）来实现的。然后智能体根据这些标准评估每个潜在行动，以确定最关键、最及时的行动方案。这种智能体能力使系统能够动态适应不断变化的环境，并有效管理受限资源。通过专注于最高优先级的项目，智能体的行为变得更加智能、稳健，并与其战略目标保持一致。

**经验法则（Rule of thumb）：** 当智能体系统必须在资源约束下自主管理多个（通常是相互冲突的）任务或目标，以在动态环境中有效运行时，使用优先级排序模式。

**可视化摘要：**

**![][image1]**

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

## 关键要点

* 优先级排序使 AI 智能体在复杂的多方面环境中有效运作。
* 智能体根据既定标准（如紧急性、重要性和依赖关系）来评估和排序任务。
* 动态重新优先级排序允许智能体根据实时变化调整其操作焦点。
* 优先级排序发生在各个层面，包括总体战略目标和即时战术决策。
* 有效的优先级排序可提高 AI 智能体效率和操作稳健性。

## 结论

总之，优先级排序模式是有效智能体 AI 的基石，使系统能够有目的地、智能地应对动态环境的复杂性。它允许智能体评估大量相互冲突的任务和目标，对在哪里集中其有限资源做出合理的决策。这种智能体能力超越了简单的任务执行，使系统能够充当主动的战略决策者。通过权衡紧急性、重要性和依赖关系等标准，智能体展现了复杂的、类似人类的推理过程。

这种智能体行为的一个关键特征是动态重新优先级排序，它赋予智能体在条件变化时实时调整关注点的自主权。如代码示例所示，智能体解释模糊的请求，自主选择和使用适当的工具，并逻辑地排列其行动以实现目标。这种自我管理工作流程的能力，正是真正的智能体系统与简单自动化脚本的区别所在。最终，掌握优先级排序对于创建能够在任何复杂的实际场景中有效、可靠运行的稳健且智能的智能体至关重要。

## 参考文献

1. Examining the Security of Artificial Intelligence in Project Management: A Case Study of AI-driven Project Scheduling and Resource Allocation in Information Systems Projects ; [https://www.irejournals.com/paper-details/1706160](https://www.irejournals.com/paper-details/1706160)
2. AI-Driven Decision Support Systems in Agile Software Project Management: Enhancing Risk Mitigation and Resource Allocation; [https://www.mdpi.com/2079-8954/13/3/208](https://www.mdpi.com/2079-8954/13/3/208)

[image1]: ../images/chapter-20/image1.png