面试06-Agent任务编排怎么处理？

Fairy要carry

194人浏览 · 2026-03-19 22:34:11

Fairy要carry · 2026-03-19 22:34:11 发布

0、背景：扁平检查表是什么？

通俗定义：

扁平 = 内存检查表没有层级、没有嵌套、所有任务都在同一级别 ；
检查表 = 只记录「任务 ID + 任务内容 + 状态」，核心作用是 核对任务完成情况 ，而非 管理任务依赖 / 层级。

TodoManager 的例子：

# 示例：self.items 的内容（扁平列表）
self.items = [
    {"id": "1", "text": "创建test.txt", "status": "completed"},
    {"id": "2", "text": "写入hello world", "status": "in_progress"},
    {"id": "3", "text": "读取文件验证", "status": "pending"}
]

无层级： 任务 1、2、3 都是平级的，没有「创建文件」是「写入内容」的父任务这种嵌套关系；
哪怕你让模型规划「先创建文件夹→再创建文件→最后写入内容」，模型也只能把这三个任务都放在同一层，无法标记「创建文件夹是创建文件的前置任务」。
无依赖： 代码只校验「同一时间只有 1 个 in_progress 任务」，但不校验「任务 A 没完成就不能开始任务 B」；
比如模型可以把「读取文件验证」标记为 in_progress，哪怕「创建文件」还是 pending（代码只会因为「两个 in_progress」报错，但不会因为「依赖未完成」报错）。

一、核心问题：平面Todo清单的致命缺陷

背景：
https://fairy-study.blog.csdn.net/article/details/159253372?spm=1001.2014.3001.5502

S03的 TodoManager 本质是内存中的扁平检查表，在复杂任务/多Agent协作场景下完全不够用，核心痛点：

无依赖关系：无法定义「任务B必须等任务A完成才能做」，模型不知道哪些任务该先做、哪些被阻塞；
无并行性：无法识别「无依赖的任务 C 和 D 可并行执行」，只能串行处理，效率极低；
无持久性：Todo只存在于内存，上下文压缩（S06）或Agent重启后就丢失，无法跨会话/跨Agent协作；
状态单一：只有「待做/进行中/完成」，无法体现「被阻塞」这种关键状态（比如任务B等A完成，此时B是「被阻塞」而非单纯「待做」）。

这些问题导致Agent在处理复杂工程任务（如「搭建数据库→写API→做认证→部署」）时，无法合理规划执行顺序、无法协作、无法断点续跑。

二、核心解决方案：基于磁盘的任务依赖图（DAG）

S07的核心是把「扁平内存清单」升级为「带依赖的持久化任务图」，核心设计：

1. 核心概念（先理解这3个词）

任务图：用「有向无环图（DAG）」描述任务关系，比如「任务1→任务2→任务4」「任务1→任务3→任务4」；
blockedBy/blocks：任务的双向依赖关系（A blocks B = B blockedBy A，即B依赖A）；
持久化：每个任务以JSON文件存在磁盘（.tasks/task_1.json），而非内存，重启/压缩都不丢。

2. 解决思路（对应问题）

问题	解决方案
依赖/并行	用`blockedBy`定义「谁阻塞我」，`blocks`定义「我阻塞谁」，支持串行+并行；
状态不完整	扩展状态为`pending→in_progress→completed`，并新增「被阻塞」的隐含状态（`blockedBy`非空则阻塞）；
无持久性	任务存在磁盘JSON文件，而非内存，跨会话/重启不丢；
无法自动解阻塞	完成任务后，自动清除所有依赖该任务的`blockedBy`列表，解除阻塞；

三、核心代码逻辑（落地实现）

1. TaskManager：任务图的核心管理器（CRUD+依赖处理）

import json
from pathlib import Path

class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)  # 确保.tasks目录存在
        self._next_id = self._max_id() + 1  # 自动生成任务ID（避免重复）

    # 私有方法：加载单个任务的内容（从JSON文件）
    def _load(self, task_id):
        f = self.dir / f"task_{task_id}.json"
        return json.loads(f.read_text())

    # 私有方法：保存任务的内容（写入JSON文件，持久化核心）
    def _save(self, task):
        f = self.dir / f"task_{task['id']}.json"
        f.write_text(json.dumps(task, indent=2))

    # 核心1：创建任务（初始状态pending，无依赖）
    def create(self, subject, description=""):
        task = {
            "id": self._next_id,
            "subject": subject,  # 任务名称（如"Setup project"）
            "status": "pending", # 初始状态
            "blockedBy": [],     # 被谁阻塞（依赖的任务ID列表）
            "blocks": [],        # 阻塞谁（被我依赖的任务ID列表）
            "owner": ""          # 可选：任务归属（多Agent时用）
        }
        self._save(task)  # 写入磁盘文件
        self._next_id += 1
        return json.dumps(task, indent=2)  # 返回创建结果

    # 核心2：更新任务（状态+依赖处理）
    def update(self, task_id, status=None, add_blocked_by=None, add_blocks=None):
        task = self._load(task_id)
        # 1. 更新状态
        if status:
            task["status"] = status
            # 关键：任务完成时，自动清除所有依赖该任务的blockedBy
            if status == "completed":
                self._clear_dependency(task_id)
        # 2. 添加依赖关系（比如任务2 blockedBy 任务1）
        if add_blocked_by:
            task["blockedBy"].extend(add_blocked_by)
        if add_blocks:
            task["blocks"].extend(add_blocks)
        self._save(task)
        return json.dumps(task, indent=2)

    # 核心3：自动解阻塞（完成任务后触发）
    def _clear_dependency(self, completed_id):
        # 遍历所有任务文件，清除包含该ID的blockedBy
        for f in self.dir.glob("task_*.json"):
            task = json.loads(f.read_text())
            if completed_id in task.get("blockedBy", []):
                task["blockedBy"].remove(completed_id)
                self._save(task)  # 写入修改后的文件

    # 辅助：列出所有任务（查看依赖图）
    def list_all(self):
        tasks = []
        for f in sorted(self.dir.glob("task_*.json")):
            tasks.append(json.loads(f.read_text()))
        return json.dumps(tasks, indent=2)

持久化： 每个任务对应一个 JSON 文件，存储在 .tasks 目录，重启不丢失；
结构化： 任务包含 ID 、状态、双向依赖（blockedBy / blocks），支持复杂依赖管理；
自动化： 任务完成时自动清除下游依赖的阻塞，无需模型手动更新；
易用性： 提供创建、更新、查询接口，模型可通过工具调用完成所有操作。

2. 工具注册（模型可调用的任务操作）

# 初始化任务管理器（持久化到.tasks目录）
TASKS = TaskManager(Path(".tasks"))

# 任务操作工具（模型可调用）
TOOL_HANDLERS = {
    # 基础工具（bash/read/write/edit/load_skill）
    # 新增4个任务工具
    "task_create": lambda **kw: TASKS.create(kw["subject"]),  # 创建任务
    "task_update": lambda **kw: TASKS.update(
        kw["task_id"], 
        status=kw.get("status"),
        add_blocked_by=kw.get("add_blocked_by"),
        add_blocks=kw.get("add_blocks")
    ),  # 更新状态 / 依赖
    "task_list":   lambda **kw: TASKS.list_all(),  # 列出所有任务（看依赖图）
    "task_get":    lambda **kw: TASKS.get(kw["task_id"]),  # 获取单个任务详情
}

# 工具定义（告诉模型怎么调用）
TOOLS.extend([
    {
        "name": "task_create",
        "description": "Create a new task with pending status",
        "input_schema": {"type": "object", "properties": {"subject": {"type": "string"}}, "required": ["subject"]}
    },
    {
        "name": "task_update",
        "description": "Update task status or dependencies",
        "input_schema": {
            "type": "object",
            "properties": {
                "task_id": {"type": "int"},
                "status": {"type": "string", "enum": ["pending", "in_progress", "completed"]},
                "add_blocked_by": {"type": "array", "items": {"type": "int"}}
            },
            "required": ["task_id"]
        }
    },
    # task_list/task_get 工具定义省略
])

四、执行流程示例（直观理解）

用户指令：「创建3个任务：Setup project（1）、Write code（2）、Write tests（3），按顺序依赖（2依赖1，3依赖2）；完成任务1后，列出所有任务看依赖变化」

步骤1：创建任务+设置依赖

模型调用task_create(subject="Setup project") → 生成.tasks/task_1.json（status=pending，blockedBy=[]）；
模型调用task_create(subject="Write code") → 生成task_2.json；
模型调用task_update(task_id=2, add_blocked_by=[1]) → 修改task_2.json的blockedBy=[1]；
模型调用task_create(subject="Write tests") → 生成task_3.json；
模型调用task_update(task_id=3, add_blocked_by=[2]) → 修改task_3.json的blockedBy=[2]；

此时任务文件内容：

task_1.json：{"id":1, "status":"pending", "blockedBy":[], "blocks":[2]}
task_2.json：{"id":2, "status":"pending", "blockedBy":[1], "blocks":[3]}
task_3.json：{"id":3, "status":"pending", "blockedBy":[2], "blocks":[]}

步骤2：完成任务1，自动解阻塞

模型调用task_update(task_id=1, status="completed")；
_clear_dependency(1) 触发：遍历所有任务，移除blockedBy中的1；
task_2.json的blockedBy变为[]（从[1]清空），状态仍为pending，但已「解除阻塞」；

步骤3：列出任务（查看结果）

模型调用task_list()，返回：

[
  {"id":1, "status":"completed", "blockedBy":[], "blocks":[2]},
  {"id":2, "status":"pending", "blockedBy":[], "blocks":[3]},  // 已解阻塞
  {"id":3, "status":"pending", "blockedBy":[2], "blocks":[]}
]

Agent Loop

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":
            return
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                try:
                    output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
                except Exception as e:
                    output = f"Error: {e}"
                print(f"> {block.name}: {str(output)[:200]}")
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
        messages.append({"role": "user", "content": results})

五、从S06到S07的核心变化（面试重点）

组成部分	S06（扁平Todo）	S07（任务图）
工具数量	5（基础+load_skill）	8（新增task_create/update/list/get）
规划模型	内存中的扁平检查表	磁盘上的依赖任务图（DAG）
任务关系	无（仅顺序）	blockedBy/blocks 双向依赖
状态体系	pending/in_progress/completed（无阻塞态）	新增「被阻塞」隐含状态（blockedBy非空）
持久性	上下文压缩/重启后丢失	磁盘文件存储，跨会话不丢
核心能力	单场次简单任务跟踪	复杂任务排序/并行/多Agent协作