第20章：Prioritization

在复杂的动态环境中，Agent 经常遇到大量潜在行动、相互冲突的目标和有限的资源。如果没有确定后续行动的定义流程，Agent 可能会遇到效率降低、操作延迟或无法达成关键目标的问题。Prioritization 模式通过使 Agent 能够根据重要性、紧迫性、依赖关系和既定标准评估和排列任务、目标或行动，从而解决了此问题。这确保了 Agent 将精力集中在最关键的任务上，从而提高有效性和目标一致性。

Prioritization 模式概述

Agent 采用 prioritization 来有效管理任务、目标和子目标，指导后续行动。当处理多种需求时，此过程有助于明智的决策，将重要或紧急的活动优先于不太关键的活动。在资源有限、时间有限且目标可能冲突的现实场景中，这一点尤其重要。

Agent prioritization 的基本方面通常涉及几个要素。首先，标准定义建立了任务评估的规则或指标。这些可能包括紧迫性（任务的时间敏感性）、重要性（对主要目标的影响）、依赖性（任务是否是其他任务的先决条件）、资源可用性（必要工具或信息的准备情况）、成本/效益分析（努力与预期结果之比）以及用于个性化 Agent 的用户偏好。其次，任务评估涉及根据这些定义的标准评估每个潜在任务，使用从简单规则到 LLM 的复杂评分或推理的方法。第三，调度或选择逻辑指的是一种算法，该算法基于评估选择最佳的下一个行动或任务序列，可能利用队列或高级规划组件。最后，动态重新优先级排序允许 Agent 随着环境变化修改优先级，例如出现新的关键事件或临近截止日期，确保 Agent 的适应性和响应能力。

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

实际应用与用例

在各种现实应用中，AI Agent 展示了 prioritization 的复杂使用，以做出及时有效的决策。

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

这些例子共同说明了 prioritization 能力对于 AI Agent 在广泛的情况范围内提高性能和决策能力至关重要。

动手代码示例

下面演示了使用 LangChain 开发项目管理 AI Agent。该 Agent 促进任务的创建、优先级排序和分配给团队成员，说明了如何将大语言模型与定制工具一起用于自动化项目管理。

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. Configuration and Setup ---
# Loads the OPENAI_API_KEY from the .env file.
load_dotenv()
# The ChatOpenAI client automatically picks up the API key from the environment.
llm = ChatOpenAI(temperature=0.5, model="gpt-4o-mini")

# --- 1. Task Management System ---
class Task(BaseModel):
    """Represents a single task in the system."""
    id: str
    description: str
    priority: Optional[str] = None  # P0, P1, P2
    assigned_to: Optional[str] = None  # Name of the worker

class SuperSimpleTaskManager:
    """An efficient and robust in-memory task manager."""
    def __init__(self):
        # Use a dictionary for O(1) lookups, updates, and deletions.
        self.tasks: Dict[str, Task] = {}
        self.next_task_id = 1

    def create_task(self, description: str) -> Task:
        """Creates and stores a new 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 created - {task_id}: {description}")
        return new_task

    def update_task(self, task_id: str, **kwargs) -> Optional[Task]:
        """Safely updates a task using Pydantic's model_copy."""
        task = self.tasks.get(task_id)
        if task:
            # Use model_copy for type-safe updates.
            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 {task_id} updated with {update_data}")
            return updated_task
        print(f"DEBUG: Task {task_id} not found for update.")
        return None

    def list_all_tasks(self) -> str:
        """Lists all tasks currently in the system."""
        if not self.tasks:
            return "No tasks in the system."
        task_strings = []
        for task in self.tasks.values():
            task_strings.append(
                f"ID: {task.id}, Desc: '{task.description}', "
                f"Priority: {task.priority or 'N/A'}, "
                f"Assigned To: {task.assigned_to or 'N/A'}"
            )
        return "Current Tasks:\n" + "\n".join(task_strings)

task_manager = SuperSimpleTaskManager()

# --- 2. Tools for the Project Manager Agent ---
# Use Pydantic models for tool arguments for better validation and clarity.
class CreateTaskArgs(BaseModel):
    description: str = Field(description="A detailed description of the task.")

class PriorityArgs(BaseModel):
    task_id: str = Field(description="The ID of the task to update, e.g., 'TASK-001'.")
    priority: str = Field(description="The priority to set. Must be one of: 'P0', 'P1', 'P2'.")

class AssignWorkerArgs(BaseModel):
    task_id: str = Field(description="The ID of the task to update, e.g., 'TASK-001'.")
    worker_name: str = Field(description="The name of the worker to assign the task to.")

def create_new_task_tool(description: str) -> str:
    """Creates a new project task with the given description."""
    task = task_manager.create_task(description)
    return f"Created task {task.id}: '{task.description}'."

def assign_priority_to_task_tool(task_id: str, priority: str) -> str:
    """Assigns a priority (P0, P1, P2) to a given task ID."""
    if priority not in ["P0", "P1", "P2"]:
        return "Invalid priority. Must be P0, P1, or P2."
    task = task_manager.update_task(task_id, priority=priority)
    return f"Assigned priority {priority} to task {task.id}." if task else f"Task {task_id} not found."

def assign_task_to_worker_tool(task_id: str, worker_name: str) -> str:
    """Assigns a task to a specific worker."""
    task = task_manager.update_task(task_id, assigned_to=worker_name)
    return f"Assigned task {task.id} to {worker_name}." if task else f"Task {task_id} not found."

# All tools the PM agent can use
pm_tools = [
    Tool(
        name="create_new_task",
        func=create_new_task_tool,
        description="Use this first to create a new task and get its ID.",
        args_schema=CreateTaskArgs
    ),
    Tool(
        name="assign_priority_to_task",
        func=assign_priority_to_task_tool,
        description="Use this to assign a priority to a task after it has been created.",
        args_schema=PriorityArgs
    ),
    Tool(
        name="assign_task_to_worker",
        func=assign_task_to_worker_tool,
        description="Use this to assign a task to a specific worker after it has been created.",
        args_schema=AssignWorkerArgs
    ),
    Tool(
        name="list_all_tasks",
        func=task_manager.list_all_tasks,
        description="Use this to list all current tasks and their status."
    ),
]

# --- 3. Project Manager Agent Definition ---
pm_prompt_template = ChatPromptTemplate.from_messages([
    ("system", """You are a focused Project Manager LLM agent. Your goal is to manage project tasks efficiently.
When you receive a new task request, follow these steps:
1. First, create the task with the given description using the `create_new_task` tool. You must do this first to get a `task_id`.
2. Next, analyze the user's request to see if a priority or an assignee is mentioned.
   - If a priority is mentioned (e.g., "urgent", "ASAP", "critical"), map it to P0. Use `assign_priority_to_task`.
   - If a worker is mentioned, use `assign_task_to_worker`.
3. If any information (priority, assignee) is missing, you must make a reasonable default assignment (e.g., assign P1 priority and assign to 'Worker A').
4. Once the task is fully processed, use `list_all_tasks` to show the final state.

Available workers: 'Worker A', 'Worker B', 'Review Team'
Priority levels: P0 (highest), P1 (medium), P2 (lowest)
"""),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# Create the agent executor
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. Simple Interaction Flow ---
async def run_simulation():
    print("--- Project Manager Simulation ---")

    # Scenario 1: Handle a new, urgent feature request
    print("\n[User Request] I need a new login system implemented ASAP. It should be assigned to Worker B.")
    await pm_agent_executor.ainvoke({"input": "Create a task to implement a new login system. It's urgent and should be assigned to Worker B."})
    print("\n" + "-"*60 + "\n")

    # Scenario 2: Handle a less urgent content update with fewer details
    print("[User Request] We need to review the marketing website content.")
    await pm_agent_executor.ainvoke({"input": "Manage a new task: Review marketing website content."})

    print("\n--- Simulation Complete ---")

# Run the simulation
if __name__ == "__main__":
    asyncio.run(run_simulation())

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

该系统采用 SuperSimpleTaskManager 类来高效地在内存中管理任务，利用字典结构进行快速数据检索。每个任务由一个 Task Pydantic 模型表示，该模型包含属性，如唯一标识符、描述性文本、可选优先级（P0、P1、P2）和可选分配者指定。内存使用量根据任务类型、worker 数量和其他因素而变化。任务管理器提供了任务创建、任务修改和检索所有任务的方法。

Agent 通过一组定义的 Tools 与任务管理器交互。这些工具促进新任务的创建、任务优先级的分配、任务给人员的分配以及所有任务的列出。每个工具都被封装以启用与 SuperSimpleTaskManager 实例的交互。Pydantic 模型用于描述工具所需的参数，从而确保数据验证。

AgentExecutor 配置了语言模型、工具集和对话内存组件，以维护上下文连续性。定义了一个特定的 ChatPromptTemplate 来指导 Agent 在项目管理角色中的行为。prompt 指示 Agent 通过创建任务开始，随后按要求分配优先级和人员，并以全面的任务列表结束。在信息缺失的情况下，prompt 中规定了默认分配，如 P1 优先级和"Worker A"。

代码结合了一个异步性质的模拟函数（run_simulation）来演示 Agent 的操作能力。该模拟执行两个不同的场景：管理具有指定人员的紧急任务，以及管理输入最少的不太紧急的任务。由于 AgentExecutor 中激活了 verbose=True，Agent 的行动和逻辑过程会输出到控制台。

一览

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

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

Rule of thumb（经验法则）：当 Agentic 系统必须在资源限制下自主管理多个（通常相互冲突的）任务或目标，以便在动态环境中有效运行时，请使用 Prioritization 模式。

视觉摘要

图1：Prioritization 设计模式

关键要点

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

结论

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

这种 agentic 行为的一个关键特征是动态重新优先级排序，它授予 Agent 随着条件变化实时调整其焦点的自主权。如代码示例所示，Agent 解释模糊的请求，自主选择并使用适当的工具，并对其行动进行逻辑排序以实现其目标。这种自我管理其工作流的能力将真正的 agentic 系统与简单的自动化脚本区分开来。最终，掌握 prioritization 对于创建强大而智能的 Agent 至关重要，这些 Agent 可以在任何复杂的现实场景中有效且可靠地运行。

参考文献

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
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