Chapter 20: Prioritization

第 20 章：优先级排序

In complex, dynamic environments, Agents frequently encounter numerous potential actions, conflicting goals, and limited resources. Without a defined process for determining the subsequent action, the agents may experience reduced efficiency, operational delays, or failures to achieve key objectives. The prioritization pattern addresses this issue by enabling agents to assess and rank tasks, objectives, or actions based on their significance, urgency, dependencies, and established criteria. This ensures the agents concentrate efforts on the most critical tasks, resulting in enhanced effectiveness and goal alignment. 在复杂、动态的环境中，智能体常常面临大量潜在行动、相互冲突的目标以及有限的资源。如果没有明确的流程来确定后续行动，智能体可能会遇到效率低下、操作延迟或无法实现关键目标的问题。优先级排序模式通过使智能体根据重要性、紧迫性、依赖关系和既定标准来评估和排序任务、目标或行动，从而解决这一挑战。这确保智能体将精力集中在最关键的任务上，从而提高有效性和目标一致性。

Prioritization Pattern Overview

优先级排序模式概述

Agents employ prioritization to effectively manage tasks, goals, and sub-goals, guiding subsequent actions. This process facilitates informed decision-making when addressing multiple demands, prioritizing vital or urgent activities over less critical ones. It is particularly relevant in real-world scenarios where resources are constrained, time is limited, and objectives may conflict. 智能体使用优先级排序来有效管理任务、目标和子目标，指导后续行动。这个过程有助于在处理多个需求时做出明智决策，优先处理重要或紧急的活动，而不是次要的活动。这在资源受限、时间有限以及目标可能冲突的实际场景中尤为重要。 The fundamental aspects of agent prioritization typically involve several elements. First, criteria definition establishes the rules or metrics for task evaluation. These may include urgency (time sensitivity of the task), importance (impact on the primary objective), dependencies (whether the task is a prerequisite for others), resource availability (readiness of necessary tools or information), cost/benefit analysis (effort versus expected outcome), and user preferences for personalized agents. Second, task evaluation involves assessing each potential task against these defined criteria, utilizing methods ranging from simple rules to complex scoring or reasoning by LLMs. Third, scheduling or selection logic refers to the algorithm that, based on the evaluations, selects the optimal next action or task sequence, potentially utilizing a queue or an advanced planning component. Finally, dynamic re-prioritization allows the agent to modify priorities as circumstances change, such as the emergence of a new critical event or an approaching deadline, ensuring agent adaptability and responsiveness. 智能体优先级排序通常包含几个核心要素。首先是标准定义，即建立任务评估的规则或指标，这些标准可能包括：紧急性（任务的时间敏感性）、重要性（对主要目标的影响）、依赖关系（该任务是否是其他任务的前提）、资源可用性（必要工具或信息的就绪状态）、成本/收益分析（投入与预期成果的对比），以及面向个性化智能体的用户偏好。其次是任务评估，即根据上述标准评估每个潜在任务，评估方法可以从简单规则到大语言模型的复杂评分或推理。第三是调度或选择逻辑，即基于评估结果选择最佳下一步行动或任务序列的算法，可能使用队列或高级规划组件。最后是动态重新优先级排序，允许智能体根据情况变化（如新关键事件出现或截止日期临近）调整优先级，确保智能体的适应性和响应能力。 Prioritization can occur at various levels: selecting an overarching objective (high-level goal prioritization), ordering steps within a plan (sub-task prioritization), or choosing the next immediate action from available options (action selection). Effective prioritization enables agents to exhibit more intelligent, efficient, and robust behavior, especially in complex, multi-objective environments. This mirrors human team organization, where managers prioritize tasks by considering input from all members. 优先级排序可以在各个层面进行：选择总体目标（高层次目标优先级排序）、在计划内排序步骤（子任务优先级排序）或从可用选项中选择下一个即时行动（行动选择）。有效的优先级排序使智能体展现更智能、更高效、更稳健的行为，特别是在复杂的多目标环境中。这反映了人类团队的组织方式，其中管理者通过考虑所有成员的输入来优先处理任务。

Practical Applications & Use Cases

实际应用和用例

In various real-world applications, AI agents demonstrate a sophisticated use of prioritization to make timely and effective decisions. 在各种实际应用中，AI 智能体展现了优先级排序的复杂运用，以做出及时有效的决策。

— 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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

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

1
2
3
4

# 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()) ```

This code implements a simple task management system using Python and LangChain, designed to simulate a project manager agent powered by a large language model.

此代码使用 Python 和 LangChain 实现了一个简单的任务管理系统，旨在模拟由大语言模型驱动的项目管理智能体。 The system employs a SuperSimpleTaskManager class to efficiently manage tasks within memory, utilizing a dictionary structure for rapid data retrieval. Each task is represented by a Task Pydantic model, which encompasses attributes such as a unique identifier, a descriptive text, an optional priority level (P0, P1, P2), and an optional assignee designation.Memory usage varies based on task type, the number of workers, and other contributing factors. The task manager provides methods for task creation, task modification, and retrieval of all tasks. 该系统采用 SuperSimpleTaskManager 类在内存中高效管理任务，利用字典结构实现快速数据检索。每个任务由 Task Pydantic 模型表示，包含唯一标识符、描述文本、可选优先级级别（P0、P1、P2）和可选的受让人指定等属性。内存使用量会根据任务类型、工作人员数量等因素而变化。任务管理器提供任务创建、任务修改和检索所有任务的方法。 The agent interacts with the task manager via a defined set of Tools. These tools facilitate the creation of new tasks, the assignment of priorities to tasks, the allocation of tasks to personnel, and the listing of all tasks. Each tool is encapsulated to enable interaction with an instance of the SuperSimpleTaskManager. Pydantic models are utilized to delineate the requisite arguments for the tools, thereby ensuring data validation. 智能体通过一组定义好的工具与任务管理器交互。这些工具支持创建新任务、为任务分配优先级、将任务分配给人员以及列出所有任务。每个工具都被封装为与 SuperSimpleTaskManager 的实例交互。Pydantic 模型用于描述工具所需的参数，从而确保数据验证。 An AgentExecutor is configured with the language model, the toolset, and a conversation memory component to maintain contextual continuity. A specific ChatPromptTemplate is defined to direct the agent’s behavior in its project management role. The prompt instructs the agent to initiate by creating a task, subsequently assigning priority and personnel as specified, and concluding with a comprehensive task list. Default assignments, such as P1 priority and ‘Worker A’, are stipulated within the prompt for instances where information is absent. AgentExecutor 配置了语言模型、工具集和对话内存组件以保持上下文连续性。定义了特定的 ChatPromptTemplate 来指导智能体在项目管理角色中的行为。提示词指示智能体先创建任务，然后根据指定分配优先级和人员，最后以完整的任务列表结束。当信息缺失时，提示词中规定了默认分配，例如 P1 优先级和 ‘Worker A’。 The code incorporates a simulation function (run_simulation) of asynchronous nature to demonstrate the agent’s operational capacity. The simulation executes two distinct scenarios: the management of an urgent task with designated personnel, and the management of a less urgent task with minimal input. The agent’s actions and logical processes are outputted to the console due to the activation of verbose=True within the AgentExecutor. 代码包含一个异步模拟函数（run_simulation）来演示智能体的协作能力。模拟执行两个不同的场景：管理带有指定人员的紧急任务，以及管理输入信息较少的非紧急任务。由于在 AgentExecutor 中启用了 verbose=True，智能体的行动和逻辑过程会输出到控制台。

At a Glance

概览

What: AI agents operating in complex environments face a multitude of potential actions, conflicting goals, and finite resources. Without a clear method to determine their next move, these agents risk becoming inefficient and ineffective. This can lead to significant operational delays or a complete failure to accomplish primary objectives. The core challenge is to manage this overwhelming number of choices to ensure the agent acts purposefully and logically. 定义（What）： 在复杂环境中运行的 AI 智能体面临大量潜在行动、相互冲突的目标和有限的资源。如果没有明确的方法来确定下一步行动，这些智能体可能会变得低效且效果不佳，甚至导致严重的操作延迟或完全无法完成主要目标。核心挑战是管理这一数量庞大的选择，确保智能体有目的地、合乎逻辑地行动。 Why: The Prioritization pattern provides a standardized solution for this problem by enabling agents to rank tasks and goals. This is achieved by establishing clear criteria such as urgency, importance, dependencies, and resource cost. The agent then evaluates each potential action against these criteria to determine the most critical and timely course of action. This Agentic capability allows the system to dynamically adapt to changing circumstances and manage constrained resources effectively. By focusing on the highest-priority items, the agent’s behavior becomes more intelligent, robust, and aligned with its strategic goals. 原因（Why）： 优先级排序模式通过使智能体对任务和目标进行排序，为这个问题提供了标准化的解决方案。这是通过建立明确的标准（如紧急性、重要性、依赖关系和资源成本）来实现的。然后智能体根据这些标准评估每个潜在行动，以确定最关键、最及时的行动方案。这种智能体能力使系统能够动态适应不断变化的环境，并有效管理受限资源。通过专注于最高优先级的项目，智能体的行为变得更加智能、稳健，并与其战略目标保持一致。 Rule of thumb: Use the Prioritization pattern when an Agentic system must autonomously manage multiple, often conflicting, tasks or goals under resource constraints to operate effectively in a dynamic environment. 经验法则（Rule of thumb）： 当智能体系统必须在资源约束下自主管理多个（通常是相互冲突的）任务或目标，以在动态环境中有效运行时，使用优先级排序模式。 Visual summary: 可视化摘要： Fig.1: Prioritization Design pattern 图 1：优先级排序设计模式

Key Takeaways

关键要点