一人即帝国

主题

04 TaskHandler：Flow + Command + Action ​

学习理念：这是整个项目代码量最大、逻辑最复杂的一篇。TaskHandler 负责执行"有步骤的业务任务"，核心是三个子系统：Flow（流程定义 + 执行器）、Command（指令解析 + 处理）、Action（业务动作执行器）。理解这三者的协作关系——Command 告诉 FlowExecutor 做什么，FlowExecutor 按 Flow 配置推进，遇到 action 就交给 ActionRunner 执行——你就掌握了任务型 Agent 的核心实现。

海外对标：Rasa Stories（对话流程定义）、LangGraph（图编排）

本节 AI 替代率：~50% | 人工干预率：~50%

角色能力范围
🤖 AI 擅长生成 Flow YAML 配置、Action 骨架代码
👤 人类需理解Flow 四种 step 类型的流转、Command 的 start_flow/set_slots/cancel_flow/resume_flow 逻辑

📌 原文说明：以下内容来自 5.设计文档/TaskHandler设计.md、TaskHandler实现_day05~07.md，代码来自 atguigu/task/ 目录。原文保留核心设计，代码以 P0-P2 标注。

📖 阅读优先级 ​

等级章节说明
🔥 必须深入Flow 四种 step 类型核心——理解 start/collect/action/end
🔥 必须深入FlowExecutor 执行逻辑外层循环+内层循环的设计
🟡 理解即可Command 处理4 种 command 的用途
🟡 理解即可Action 系统知道内置+自定义的分离设计
🟢 了解即可ActionRunner Builder了解 Action 是如何被构造的

一、TaskHandler 整体架构 ​

TaskHandler.process(TurnPlan)
    ↓
CommandProcessor 处理 TaskTurnPlan.commands
    ↓
    每个 command 执行对应的操作
    ↓
        start_flow → FlowExecutor 启动一个新流程
        set_slots  → 填充当前流程的槽位
        cancel_flow → 取消当前流程
        resume_flow → 恢复挂起的流程
    ↓
FlowExecutor 按 Flow 配置一步步推进
    ↓
    start_step → 流程起点
    collect_step → 收集用户槽位信息
    action_step → 执行 Action（如查订单、查物流）
    end_step → 流程结束
    ↓
ActionRunner 执行业务 Action
    ├── 内置 Action（action_response, action_listen）
    └── 自定义 Action（lookup_order, lookup_logistics, recommend_product）

二、Flow 配置语法 ​

🔥【P0 必须理解】 Flow 用 YAML 定义，包括 slots（槽位定义）和 flows（流程步骤）。

2.1 slots 定义 ​

yaml
slots:
  order_number:
    type: text
    label: 订单号
  refund_reason:
    type: text
    label: 退款原因

2.2 flows 定义 ​

yaml
flows:
  refund_request:
    name: 退款申请
    description: 帮用户提交简单的退款申请
    steps:
      - id: start
        type: start
        next: ask_order_number

      - id: ask_order_number
        type: collect
        slot_name: order_number
        response:
          text: "请告诉我你的订单号。"
        next: ask_refund_reason

      - id: ask_refund_reason
        type: collect
        slot_name: refund_reason
        response:
          text: "请简单说一下退款原因。"
        next: refund_submitted

      - id: refund_submitted
        type: action
        action: action_response
        args:
          text: "订单{{ slots.order_number }}的退款申请已提交，原因：{{ slots.refund_reason }}。"
        next: end

      - id: end
        type: end

2.3 四种 step 类型 ​

类型作用关键字段说明
start流程起点next固定行为，不需要额外配置
collect收集用户信息slot_name、response向用户提问，等待回答填充槽位
action执行业务动作action、args调用 Action（内置或自定义），可传参
end流程结束无关闭当前 flow

三、Flow 模型定义 ​

🟡【P1 看注释就行】 Flow 模型对应 YAML 配置的结构：

python
@dataclass
class SlotInfo:
    name: str
    type: str
    label: str

@dataclass
class Step:
    id: str
    type: str                # start / collect / action / end
    slot_name: str = ""      # collect 类型用到
    response: dict = field(default_factory=dict)
    action: str = ""         # action 类型用到
    args: dict = field(default_factory=dict)
    next: str | list[str] = ""

@dataclass
class Flow:
    id: str
    name: str
    description: str = ""
    slots: list[SlotInfo] = field(default_factory=list)
    steps: list[Step] = field(default_factory=list)

四、Command 处理 ​

🟡【P1 看注释就行】 TurnPlan 中的 commands 会被 CommandProcessor 解析和执行。

4.1 Command 模型 ​

python
@dataclass
class Command:
    command: str              # start_flow / set_slots / cancel_flow / resume_flow
    flow: str = ""           # 流程 ID（start_flow / resume_flow 时需要）
    slots: dict = field(default_factory=dict)  # 槽位键值对（set_slots 时需要）

4.2 四种 Command ​

Command触发场景执行逻辑
start_flow用户说"我要退款"创建新的 TaskContext，启动 FlowExecutor
set_slots用户说"我的订单号是 123"把值填入当前 active_task 的 slots
cancel_flow用户说"算了不办了"清除当前 active_task
resume_flow用户说"继续之前的退款"把 paused_tasks 队首的移到 active_task

五、FlowExecutor——执行引擎 ​

🔥【P0 必须理解】 FlowExecutor 包含两层循环：

  • 外层循环（run_task）：当当前步骤有 next 时，继续推进
  • 内层循环（advance_until_action）：推进到第一个 action 或 end 才停下

5.1 执行流程示意 ​

用户输入 → CommandProcessor 处理 commands
    ↓
start_flow "refund_request"  → 创建 TaskContext，step_id = "start"
    ↓
FlowExecutor.run_task()
    ↓
外层循环: 当前 step = "start"
    ↓ 有 next → 前进到 "ask_order_number"
内层循环: 当前 step = "ask_order_number" (type=collect)
    ↓ collect 需要用户输入 → 暂停，等待用户回复
    ↓ 回复内容被 set_slots 接收 → 槽位填充完毕
    ↓ 前进到 "ask_refund_reason"
    ↓ 同样 collect → 等待
    ↓ 前进到 "refund_submitted" (type=action)
    ↓ ActionRunner 执行 action_response
    ↓ 前进到 "end" (type=end)
    ↓ 流程结束，清除 active_task

5.2 核心代码逻辑 ​

python
class FlowExecutor:
    async def run_task(self, state: DialogueState) -> BotMessage | None:
        """外层循环：只要还有 next 就继续推进"""
        while self._has_next_step(state):
            result = await self._advance_until_action(state)
            if result is not None:
                return result
        return None

    async def _advance_until_action(self, state: DialogueState) -> BotMessage | None:
        """内层循环：推进到第一个 action 或 end 才停下"""
        flow = self._get_flow(state)
        while True:
            step = self._get_current_step(flow, state)
            
            if step.type == "start":
                self._handle_start_step(state, flow, step)
                
            elif step.type == "collect":
                if self._is_slot_filled(state, step):
                    self._move_to_next_step(state, flow, step)
                else:
                    # 槽位未填 → 向用户提问 → 暂停
                    return self._handle_collect_step(state, step)
                
            elif step.type == "action":
                # 执行 Action → 返回回复
                return await self._handle_action_step(state, step)
                
            elif step.type == "end":
                self._handle_end_step(state, flow)
                return None

六、Action 系统 ​

🟡【理解即可】 Action 分为内置（系统自带）和自定义（业务相关），由 ActionRunner 统一执行。

6.1 Action 基类 ​

python
class Action:
    name: str  # Action 名称，对应 flow 配置中的 action 字段
    
    async def run(self, args: dict, state: DialogueState) -> BotMessage:
        """执行动作，返回机器人回复"""
        ...

6.2 内置 Action ​

Action作用参数
action_response用 Jinja2 模板生成回复text：模板字符串，可引用 slots.xxx
action_listen等待用户输入无

6.3 自定义 Action ​

Action作用实现
lookup_order_status查订单状态调用电商 API GET /orders/{id}/status
lookup_logistics查物流调用电商 API GET /orders/{id}/logistics
recommend_similar_products推荐相似商品调用电商 API + 模拟推荐逻辑

6.4 ActionRunner ​

python
class ActionRunner:
    def __init__(self):
        self._builtin_actions: dict[str, Action] = {}    # 内置
        self._custom_actions: dict[str, Action] = {}     # 自定义

    async def run_action(self, action_name: str, args: dict, state: DialogueState) -> BotMessage:
        # 先从自定义找，再从内置找
        action = self._custom_actions.get(action_name) or self._builtin_actions.get(action_name)
        return await action.run(args, state)

七、企业痛点-方案映射 ​

痛点传统方案AI Agent 方案
业务审批流程变更频繁改后端代码+部署修改 YAML 配置即可
需要对接多个业务 API硬编码调用链自定义 Action 独立开发注册
用户中途改变主意（中断/切换任务）人工客服处理active_task + paused_tasks 挂起/恢复

八、本阶段文件索引 ​

优先级文件路径说明
🔥 P0task/flow/executor.pyatguigu/task/flow/executor.pyFlowExecutor 执行引擎
🔥 P0task/flow/steps.pyatguigu/task/flow/steps.py四种 step 实现
🔥 P0task/command/processor.pyatguigu/task/command/processor.pyCommand 处理器
🟡 P1task/flow/models.pyatguigu/task/flow/models.pyFlow/Slot/Step 模型
🟡 P1task/flow/loader.pyatguigu/task/flow/loader.pyYAML -> Flow 模型加载
🟡 P1task/action/runner.pyatguigu/task/action/runner.pyActionRunner
🟡 P1task/action/builtin/action_response.pyatguigu/task/action/builtin/action_response.py内置 Action
🟢 P2task/action/custom/lookup_order_status.pyatguigu/task/action/custom/自定义 Action 示例
🟢 P2flow_config/user_flows.ymlflow_config/user_flows.ymlFlow YAML 配置

OPC 超级个体实战指南

© 2026 OPC Guide