# 1. 实现思路 `TaskHandler` 是任务轨道的入口。它不负责理解用户自然语言，也不负责读写数据库。它只接收已经解析好的 `commands` 和当前 `DialogueState`，然后推进任务流程。 commands包括: `start_flow` `cancel_flow` `resume_flow` `set_slots` 它内部主要依赖三个对象： | 对象 | 职责 | | --- | --- | | `CommandProcessor` | 处理命令，修改 `DialogueState`。 | | `FlowExecutor` | 根据当前状态推进 flow。 | | `ActionRunner` | 执行 flow 中声明的 action。 | 因此，本章的实现顺序也围绕这条链路展开： ```text 1. 先实现 flow 的定义和加载 2. 再实现 command 的定义和解析 3. 然后实现 CommandProcessor 4. 接着实现 ActionRunner 5. 再实现 FlowExecutor 6. 最后实现 TaskHandler ``` # 2. Flow定义与加载 ## 2.1 模型定义 Flow具体配置示例如下：直接从资料中拷贝flow_config文件夹拷贝到项目根目录与atguigu包并列 ```yaml slots: order_number: type: text label: 订单号 description: 用户要申请退款的订单号 refund_reason: type: text label: 退款原因 description: 用户申请退款的原因 flows: refund_request: name: 退款申请 description: 收集订单号和退款原因，提交退款申请。 slots: - order_number - refund_reason steps: - id: start type: start description: 流程入口 next: ask_order_number - id: ask_order_number type: collect description: 收集订单号 slot_name: order_number response: mode: static text: "请告诉我你的订单号。" validation: condition: "slots.get('order_number')" failure_response: mode: static text: "订单号不能为空，请重新输入。" next: ask_refund_reason - id: ask_refund_reason type: collect description: 收集退款原因 slot_name: refund_reason response: mode: static text: "请简单说一下退款原因。" next: - if: "slots.get('refund_reason')" then: submit_refund - else: ask_refund_reason - id: submit_refund type: action description: 提交退款申请并回复用户 action: action_response args: mode: static text: "好的，订单{{ slots.order_number }}的退款申请已提交。" next: end - id: end type: end description: 流程结束 next: [] ``` 与之对应的类图如下：

各模型具体定义如下：在atguigu包下创建task/flow包中添加模块 `links.py`： ```python from dataclasses import dataclass @dataclass(slots=True) class FlowStepLink: target: str @dataclass(slots=True) class StaticLink(FlowStepLink): pass @dataclass(slots=True) class ConditionalLink(FlowStepLink): condition: str @dataclass(slots=True) class FallbackLink(FlowStepLink): pass ``` `steps.py`： ```python from dataclasses import dataclass, field from enum import Enum from typing import Dict, Any, List from atguigu.task.flow.links import FlowStepLink class FlowStepType(str, Enum): START = "start" ACTION = "action" COLLECT = "collect" END = "end" @dataclass(slots=True) class ResponseDefinition: mode: str = "static" text: str | None = None prompt: str | None = None @dataclass(slots=True) class SlotValidation: condition: str | None = None failure_response: ResponseDefinition | None = None @dataclass(slots=True) class FlowStep: id: str type: FlowStepType next: List[FlowStepLink] = field(default_factory=list) description: str = "" @dataclass(slots=True) class StartFlowStep(FlowStep): pass @dataclass(slots=True) class ActionFlowStep(FlowStep): action: str = "" args: Dict[str, Any] = field(default_factory=dict) @dataclass(slots=True) class CollectSlotStep(FlowStep): slot_name: str = "" response: ResponseDefinition = field(default_factory=ResponseDefinition) validation: SlotValidation | None = None @dataclass(slots=True) class EndFlowStep(FlowStep): pass ``` `models.py`： ```python from dataclasses import dataclass, field from typing import List, Dict from atguigu.task.flow.steps import FlowStep @dataclass(slots=True) class FlowSlot: name: str type: str = "any" label: str = "" description: str = "" @dataclass(slots=True) class Flow: id: str description: str = "" steps: List[FlowStep] = field(default_factory=list) slots: List[FlowSlot] = field(default_factory=list) name: str | None = None @dataclass(slots=True) class FlowsList: flows: List[Flow] = field(default_factory=list) slots: Dict[str, FlowSlot] = field(default_factory=dict) ``` ## 2.2 加载定义好的flow.yml 要求实现一个FlowLoader，支持从多个flow.yml文件中加载配置，得到FlowsList对象。 ### 2.2.1 加载 user_flows.yml -slots 在task/flow包下创建loader.py模块 ```python class FlowLoader: """ 加载文件 """ def load(self, path: Path) -> FlowsList: # 读取文件 with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) print(data) slots = self._load_slots(data.get("slots", {})) print(slots) # return data def _load_slots(self, slots_data: dict[str, dict]) -> dict[str, FlowSlot]: # 定义一个字典 slots = {} for slot_name, slots_info in slots_data.items(): # 因为yml文件格式缺少name，所以需要手动添加。 slots[slot_name] = FlowSlot(**slots_info, name=slot_name) return slots if __name__ == '__main__': flow_path = Path(__file__).parents[3] / "flow_config" / "user_flows.yml" loader = FlowLoader() loader.load(flow_path) ``` ### 2.2.2 加载 user_flows.yml -flows 1. 在load() 函数中添加一行代码 ```python def load(self, path: Path) -> FlowsList: # 读取文件 with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) # 如果为空，则个诶一个默认的字典。如果不为空，则返回一个字典。 # 不能使用data["slots"] slots = self._load_slots(data.get("slots", {})) # 加载flows; flows = self._load_flows(data.get("flows", {}), slots) print(flows) # 返回数据 return FlowsList(flows=flows, slots=slots) ``` 2. 在loader.py 模块中添加对应_load_flows() 函数 ```python def _load_flows(self, flows_data: dict[str, dict], slots: dict[str, FlowSlot]) -> list[Flow]: """ 从配置数据中加载所有流程定义 Args: flows_data: 流程配置数据字典，键为流程ID，值为流程配置信息 slots: 预加载的槽位定义字典，键为槽位名称，值为FlowSlot对象 Returns: 解析后的Flow对象列表 """ flows: list[Flow] = [] for flow_id, flow_info in flows_data.items(): # 将步骤配置数据转换为FlowStep对象列表 steps: list[FlowStep] = [ FlowStep.from_dict(step_info) for step_info in flow_info["steps"] ] # 收集流程中所有需要收集的槽位定义 flow_slots: list[FlowSlot] = [] for step in steps: if isinstance(step, CollectSlotStep): flow_slots.append(slots[step.slot_name]) # 创建Flow对象并添加到列表 flow = Flow( id=flow_id, name=flow_info.get("name"), description=flow_info.get("description"), steps=steps, slots=flow_slots ) flows.append(flow) return flows ``` 3. 在FlowStep添加对应的函数 ```python @dataclass(slots=True) class FlowStep: id: str type: FlowStepType next: List[FlowStepLink] = field(default_factory=list) description: str = "" @classmethod def from_dict(cls, step_info: dict[str, Any]) -> "FlowStep": # 获取type: step_type = step_info['type'] # 声明一个map 映射 clz = STEP_TYPE_TO_CLASS[step_type] # 调用from_dict函数; return clz.from_dict(step_info) # 定义好公共的返回结构 @staticmethod def base_fields(step_info: dict[str, Any]) -> dict[str, Any]: return { "id": step_info["id"], "type": FlowStepType(step_info["type"]), "next": FlowStep.build_links(step_info["next"]), "description": step_info.get("description", "") } @staticmethod def build_links(next_data: str | list) -> list[FlowStepLink]: if isinstance(next_data, str): return [StaticLink(target=next_data)] else: links = [] # lind_data:属于一个字典结构 # next: # - if: "slots.get('refund_reason')" # then: submit_refund # - else: ask_refund_reason for link_data in next_data: if "if" in link_data: links.append(ConditionalLink( target=link_data["then"], condition=link_data["if"] )) else: links.append(FallbackLink(target=link_data["else"])) return links @dataclass(slots=True) class StartFlowStep(FlowStep): pass @classmethod def from_dict(cls, step_info: dict[str, Any]) -> "StartFlowStep": # 需要结构 return cls(**FlowStep.base_fields(step_info)) @dataclass(slots=True) class ActionFlowStep(FlowStep): action: str = "" args: Dict[str, Any] = field(default_factory=dict) @classmethod def from_dict(cls, step_info: dict[str, Any]) -> "ActionFlowStep": return cls(**FlowStep.base_fields(step_info), action=step_info["action"], args=step_info.get("args", {})) @dataclass(slots=True) class CollectSlotStep(FlowStep): slot_name: str = "" response: ResponseDefinition = field(default_factory=ResponseDefinition) validation: SlotValidation | None = None @classmethod def from_dict(cls, step_info: dict[str, Any]) -> "CollectSlotStep": return cls(**FlowStep.base_fields(step_info), slot_name=step_info["slot_name"], response=ResponseDefinition(**step_info["response"]), validation=SlotValidation(condition=step_info["validation"]['condition'], failure_response=ResponseDefinition(**step_info["validation"][ "failure_response"])) if "validation" in step_info else None ) @dataclass(slots=True) class EndFlowStep(FlowStep): pass @classmethod def from_dict(cls, step_info: dict[str, Any]) -> "EndFlowStep": return cls(**FlowStep.base_fields(step_info)) # 定义一个字典; STEP_TYPE_TO_CLASS = { "start": StartFlowStep, "action": ActionFlowStep, "collect": CollectSlotStep, "end": EndFlowStep, } ``` ### 2.2.3 加载多文件函数 ```python class FlowLoader: def load_many(self, paths: list[Path]) -> FlowsList: """ 从多个配置文件路径批量加载流程定义 Args: paths: 配置文件路径列表 Returns: 合并后的FlowsList对象，包含所有流程和槽位定义 """ flows: list[Flow] = [] slots: dict[str, FlowSlot] = {} # 遍历所有配置文件，加载并合并流程和槽位数据 for path in paths: flows_list: FlowsList = self.load(path) flows.extend(flows_list.flows) slots.update(flows_list.slots) return FlowsList(flows=flows, slots=slots) if __name__ == '__main__': flow_path = Path(__file__).parents[3] / "flow_config" / "user_flows.yml" system_flow_path = Path(__file__).parents[3] / "flow_config" / "system_flows.yml" loader = FlowLoader() flows_list = loader.load_many([flow_path, system_flow_path]) print(flows_list) ``` # 3. Command定义与解析 ## 3.1 模型定义 Command由TurnPlanner调用LLM输出，其具体格式如下：启动任务： ```json {"command": "start_flow", "flow": "refund_request"} ``` 写入槽位： ```json {"command": "set_slots", "slots": {"order_number": "10001"}} ``` 取消任务： ```json {"command": "cancel_flow"} ``` 恢复任务： ```json {"command": "resume_flow", "flow": "refund_request"} ``` ## 3.2 解析对应的模型定义如下: 在task下创建一个command包，再创建一个models.py 要求能够由LLM输出的JSON格式命令解析为对应的Command对象。解析命令 ```python from dataclasses import dataclass from typing import Any @dataclass class Command: command: str @classmethod def from_dict(cls, data: dict[str, Any]) -> "Command": clz = COMMAND_NAME_TO_CASE[data["command"]] return clz(**data) @dataclass class StartFlowCommand(Command): flow: str @dataclass class SetSlotsCommand(Command): slots: dict[str, Any] @dataclass class CancelFlowCommand(Command): pass @dataclass class ResumeFlowCommand(Command): flow: str COMMAND_NAME_TO_CASE = { "start_flow": StartFlowCommand, "set_slots": SetSlotsCommand, "cancel_flow": CancelFlowCommand, "resume_flow": ResumeFlowCommand } if __name__ == '__main__': command = {"command": "set_slots", "slots": {"order_number": "10001"}} print(Command.from_dict(command)) ``` # 4. CommandProcessor ## 4.1 概述 `CommandProcessor` 的职责，就是把 `Command` 应用到 `DialogueState` 上。 `CommandProcessor` 主要修改三类状态： | 状态 | 说明 | | --- | --- | | `active_task` | 当前正在处理的用户任务上下文。 | | `paused_tasks` | 被打断后暂停的用户任务上下文。 | | `active_system_flow` | 当前需要优先执行的系统流程上下文。 | 入口方法只做一件事：遍历 commands。还是在command包下创建processor.py ```python from atguigu.domain.state import DialogueState from atguigu.task.command.models import Command, StartFlowCommand, SetSlotsCommand, CancelFlowCommand, ResumeFlowCommand from atguigu.task.flow.models import FlowsList class CommandProcessor: def run( self, commands: list[Command], state: DialogueState, flows: FlowsList, ) -> None: for command in commands: self._apply(command, state, flows) def _apply( self, command: Command, state: DialogueState, flows: FlowsList, ) -> None: if isinstance(command, StartFlowCommand): self._handle_start_flow(command, state, flows) elif isinstance(command, SetSlotsCommand): self._handle_set_slots(command, state) elif isinstance(command, CancelFlowCommand): self._handle_cancel_flow(state, flows) elif isinstance(command, ResumeFlowCommand): self._handle_resume_flow(command, state, flows) ``` 注意：`_apply()` 根据 Command 类型分发： ## 4.2 具体处理逻辑 ### 4.2.1 start_flow `start_flow` 表示启动一个业务任务。 ```python StartFlowCommand(command="start_flow", flow="refund_request") ``` 处理流程如下：

需要的 SystemContext：将这两个类放入到domain包下的contexts.py 中，此时需要将原来的to_dict()，from_dict() 两个函数进行修改 ```python @dataclass class StartedSystemContext(SystemContext): started_flow_id: str = "" started_flow_name: str = "" @dataclass class InterruptedSystemContext(SystemContext): interrupted_flow_id: str = "" interrupted_flow_name: str = "" started_flow_id: str = "" started_flow_name: str = "" FLOW_ID_TO_CONTEXT_CLASS={ "system_task_started": StartedSystemContext, "system_task_interrupted": InterruptedSystemContext } ``` 修改之后： ```python @dataclass class SystemContext: """系统流程执行快照（对应系统发起的交互，如收集信息、确认、纠错等）。""" flow_id: str step_id: str | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, data: dict[str, Any]) -> "SystemContext": clz = FLOW_ID_TO_CONTEXT_CLASS[data["flow_id"]] return clz(**data) ``` #### 4.2.1.1 代码实现 processor.py模块代码 ```python def _handle_start_flow(self, command: StartFlowCommand, state: DialogueState, flows: FlowsList): """ 启动工作流 :param command: :param state: :param flows: :return: """ # 清空数据 state.end_system_task() # 获取目标flow target_flow: Flow = flows.get_flow_by_id(command.flow) # 判断当前是否有正在执行的task active_task = state.active_task if active_task: # 暂停当前正在执行的task state.interrupt_activate_task() # 启动薪任务 state.start_task( TaskContext( flow_id=target_flow.id, step_id=target_flow.start_step().id, ) ) # 启动系统任务：system_task_started state.start_system_task(InterruptedSystemContext( flow_id="system_task_interrupted", step_id=flows.get_flow_by_id("system_task_interrupted").start_step().id, interrupted_flow_id=active_task.flow_id, interrupted_flow_name=flows.get_flow_by_id(active_task.flow_id).name, started_flow_id=target_flow.id, started_flow_name=target_flow.name )) else: state.start_system_task( TaskContext( flow_id=target_flow.id, step_id=target_flow.start_step().id, ) ) state.start_system_task(StartedSystemContext( flow_id="system_task_started", step_id=flows.get_flow_by_id("system_task_started").start_step().id, started_flow_id=target_flow.id, started_flow_name=target_flow.name )) ``` models.py ```python @dataclass(slots=True) class FlowsList: flows: List[Flow] = field(default_factory=list) slots: Dict[str, FlowSlot] = field(default_factory=dict) def get_flow_by_id(self, flow_id: str) -> Flow: """ 根据id获取flow :param flow_id: :return: """ for flow in self.flows: if flow.id == flow_id: return flow return None @dataclass(slots=True) class Flow: id: str description: str = "" steps: List[FlowStep] = field(default_factory=list) slots: List[FlowSlot] = field(default_factory=list) name: str | None = None def start_step(self) -> FlowStep | None: for step in self.steps: if step.type == FlowStepType.START: return step return None ``` state.py ```python def start_system_task(self, system_context: SystemContext): """ 启动系统任务 :param system_context: :return: """ self.active_system_flow = system_context # 清空数据 def end_system_task(self): """ 关闭当前的系统任务 :return: """ self.active_system_flow = None # 暂停当前正在执行的task def interrupt_activate_task(self): """ 中断当前任务 :return: """ self.paused_tasks.append(self.active_task) self.active_task = None def start_task(self, task_context: TaskContext): """ 启动任务 :param task_context: :return: """ self.active_task = task_context ``` ### 4.2.2 set_slots `set_slots` 表示用户补充了业务字段。 ```python SetSlotsCommand( command="set_slots", slots={"order_number": "10001"}, ) ``` 处理流程如下：

#### 4.2.2.1 代码实现 ```python def _handle_set_slots(self, command:SetSlotsCommand, state: DialogueState): """ 设置slots command：{"command": "set_slots", "slots": {"order_number": "10001"}} :param command: :param state: :return: """ if state.active_task: state.set_slots(command.slots) ``` state.py ```python def set_slots(self, slots: dict[str, Any]): """ 更新槽位 :param slots: :return: """ self.active_task.slots.update(slots) ``` ### 4.2.3 cancel_flow `cancel_flow` 表示取消当前任务。 ```python CancelFlowCommand(command="cancel_flow") ``` 处理流程如下：

#### 4.2.3.1 代码实现需要的 SystemContext：放入contents.py中 ```python @dataclass class CanceledSystemContext(SystemContext): canceled_flow_id: str = "" canceled_flow_name: str = "" # 补充一下map映射关系 FLOW_ID_TO_CONTEXT_CLASS={ "system_task_started": StartedSystemContext, "system_task_interrupted": InterruptedSystemContext, "system_task_canceled": CanceledSystemContext, } ``` ```python def _handle_cancel_flow(self, state: DialogueState, flows: FlowsList): # 获取到当前任务 active_task = state.active_task # 根据id获取到Flow target_flow = flows.get_flow_by_id(active_task.flow_id) state.cancel_task() # 启动系统任务 state.start_system_task( CanceledSystemContext( flow_id="system_task_canceled", step_id=flows.get_flow_by_id("system_task_canceled").start_step().id, canceled_flow_id=target_flow.id, canceled_flow_name=target_flow.name ) ) ``` state.py ```python def cancel_task(self): self.active_task = None ``` ### 4.2.4 resume_flow `resume_task` 表示恢复指定暂停任务。 ```python ResumeTaskCommand( command="resume_task", flow="order_status_query", ) ``` 处理流程如下：

#### 4.2.4.1 代码实现需要在contexts.py 添加 ```python @dataclass class ResumedSystemContext(SystemContext): resumed_flow_id: str = "" resumed_flow_name: str = "" FLOW_ID_TO_CONTEXT_CLASS={ "system_task_started": StartedSystemContext, "system_task_interrupted": InterruptedSystemContext, "system_task_canceled": CanceledSystemContext, "system_task_resumed": ResumedSystemContext, } ``` processor.py ```python def _handle_resume_flow(self, command:ResumeFlowCommand, state: DialogueState, flows: FlowsList): active_task: TaskContext = state.active_task target_flow:Flow = flows.get_flow_by_id(command.flow) if active_task: # 当前活跃的工作流Id if active_task.flow_id == target_flow.id: return else: # 暂停当前正在执行的task state.interrupt_activate_task() # 恢复目标任务 state.resume_task(target_flow.id) # 启动系统任务 state.start_system_task( InterruptedSystemContext( flow_id="system_task_interrupted", step_id=flows.get_flow_by_id("system_task_interrupted").start_step().id, interrupted_flow_id=active_task.flow_id, interrupted_flow_name=flows.get_flow_by_id(active_task.flow_id).name, started_flow_id=target_flow.id, started_flow_name=target_flow.name ) ) else: # 恢复目标任务 state.resume_task(target_flow.id) # 启动系统任务 state.start_system_task( ResumedSystemContext( flow_id="system_task_resumed", step_id=flows.get_flow_by_id("system_task_resumed").start_step().id, resumed_flow_id=target_flow.id, resumed_flow_name=target_flow.name ) ) ``` state.py ```python def resume_task(self, flow_id:str): """ 回复目标任务 :param flow_id: :return: """ for paused_task in self.paused_tasks: if paused_task.flow_id == flow_id: self.active_task = paused_task # 删除之后后续就不需要遍历了。 self.paused_tasks.remove(paused_task) return ``` # 5. ActionRunner ## 5.1 概述 `ActionRunner` 负责执行 flow step 中声明的 action。它接收 `FlowExecutor` 返回的 `ActionCall`，然后执行对应的动作。 `ActionCall` 的结构如下： ```python @dataclass class ActionCall: action_name: str action_kwargs: dict[str, Any] = field(default_factory=dict) ``` 其中： | 字段 | 说明 | | --------------- | ---------------------- | | `action_name` | 要执行的 action 名称。 | | `action_kwargs` | 传给 action 的参数。 | `ActionRunner` 的整体流程如下：

相关概念如下： | 概念 | 说明 | | --- | --- | | `ActionRunner` | action 执行入口。 | | `ActionRegistry` | action 注册表。 | | `Action` | 具体动作的基类。 | | `ActionCall` | 一次 action 调用请求。 | | `ActionResult` | action 的执行结果。 | ## 5.2 Action `Action` 是所有运行时动作的基类。在task包下创建一个action包创建一个base.py 将代码复制进去每个 action 都有一个唯一的 `name`。 ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any from atguigu.domain.messages import BotMessage from atguigu.domain.state import DialogueState class Action(ABC): name: str @abstractmethod async def run( self, state: DialogueState, action_kwargs: dict[str, Any], ) -> ActionResult: pass ``` 新增 action 时，只需要继承 `Action`，实现 `run()` 方法，并注册到 `ActionRegistry`。 ## 5.3 ActionResult `ActionResult` 是`Action.run()` 方法返回值类型。这个类也在base.py模块中 ```python @dataclass class ActionResult: messages: list[BotMessage] = field(default_factory=list) slot_updates: dict[str, Any] = field(default_factory=dict) ``` `ActionResult` 包含两类结果： | 字段 | 说明 | | --- | --- | | `messages` | 本次 action 生成的机器人消息。 | | `slot_updates` | 本次 action 要写回的槽位。 | ## 5.4 ActionRegistry `ActionRegistry` 是 action 注册表。在action包下创建registry.py 它负责维护： ```text action_name -> Action 实例 ``` 源码如下： ```python from typing import Dict from atguigu.task.action.base import Action class ActionRegistry: """Resolver for runtime actions.""" def __init__(self) -> None: self._actions: Dict[str, Action] = {} def register(self, action: Action) -> None: self._actions[action.name] = action def get(self, name: str) -> Action: if name not in self._actions: raise KeyError(f"Unknown action '{name}'.") return self._actions[name] ``` 方法说明： | 方法 | 说明 | | --- | --- | | `register(action)` | 注册一个 action。 | | `get(name)` | 根据名称取得 action。 | 使用方式： ```python registry = ActionRegistry() registry.register(ActionResponse()) registry.register(ActionListen()) ``` `ActionRunner` 执行 action 时，会从 `ActionRegistry` 中查找。 ## 5.5 ActionRunner 在action包下创建runner.py模块 ```python from dataclasses import field, dataclass from typing import Any from atguigu.domain.state import DialogueState from atguigu.task.action.base import ActionResult from atguigu.task.action.registry import ActionRegistry @dataclass class ActionCall: action_name: str action_kwargs: dict[str, Any] = field(default_factory=dict) class ActionRunner: """Execute actions requested by the flow executor.""" def __init__( self, registry: ActionRegistry, ) -> None: self.registry = registry async def run( self, action_call: ActionCall, state: DialogueState, ) -> ActionResult: action_name = action_call.action_name action = self.registry.get(action_name) return await action.run(state, action_call.action_kwargs) ``` `ActionRunner` 不关心 action 的内部逻辑。 `ActionCall` ```python @dataclass class ActionCall: action_name: str action_kwargs: dict[str, Any] = field(default_factory=dict) ``` 其中： | 字段 | 说明 | | --------------- | ---------------------- | | `action_name` | 要执行的 action 名称。 | | `action_kwargs` | 传给 action 的参数。 | 它只负责： | 步骤 | 说明 | | --- | --- | | 查找 | 根据 `action_call.action_name` 找到 `Action`。 | | 执行 | 调用 `Action.run()`。 | | 返回 | 把 `ActionResult` 交回 `FlowExecutor`。 | ## 5.6 内置 Action ### 5.6.1 action_response #### 5.6.1.1 概述 `action_response` 用于生成客服回复。内置action在builtin包中它读取 flow step 中的 `args`，结合当前 `DialogueState`，最终返回一个包含 `BotMessage` 的 `ActionResult`。 ```yaml type: action action: action_response args: mode: static text: "订单{{ slots.order_number }}正在配送中。" ``` 整体过程如下：

最终返回： ```python ActionResult(messages=[BotMessage(text=result)]) ``` #### 5.6.1.2 参数说明 `action_response` 主要读取三个参数： | 参数 | 说明 | | --- | --- | | `mode` | 回复生成模式。 | | `text` | 回复文本模板。 | | `prompt` | LLM 指令模板。 | ##### 5.6.1.2.1 mode `mode` 决定回复如何生成。 | mode | 说明 | | --- | --- | | `static` | 只渲染 `text`，不调用 LLM。 | | `rephrase` | 先渲染 `text`，再让 LLM 改写。 | | `generate` | 直接使用 `prompt` 让 LLM 生成。 | `static` 示例： ```yaml type: action action: action_response args: mode: static text: "订单{{ slots.order_number }}当前状态是{{ slots.order_status }}我会继续帮你跟进。" ``` `rephrase` 示例： ```yaml type: action action: action_response args: mode: rephrase text: "订单{{ slots.order_number }}的退款申请已经提交。" prompt: | 你是一个中文电商客服助手，语气自然、友好、简洁。请结合对话上下文，把下面的建议回复改写得更自然，但不要改变含义。对话历史： {{ history }} 用户最后一句： {{ user_message }} 建议回复：{{ current_response }} ``` `generate` 示例： ```yaml type: action action: action_response args: mode: generate prompt: | 你是一个中文电商客服助手，语气自然、友好、简洁。请根据对话上下文和用户最后一句，生成一句客服回复。对话历史：{{ history }} 用户最后一句：{{ user_message }} ``` ##### 5.6.1.2.2 text `text` 是回复文本模板。它用于生成最终可能展示给用户的回复。在 `static` 模式下，`text` 渲染后的结果就是最终回复。在 `rephrase` 模式下，`text` 会先被渲染成 `current_response`，再交给 LLM 改写。 `text` 支持 Jinja2 模板渲染。常用参数： | 参数 | 说明 | | --- | --- | | `slots` | 当前业务任务槽位。 | | `context` | 当前系统流程上下文。 | 示例： ```yaml text: "订单{{ slots.order_number }}当前状态是：{{ slots.order_status }}。" ``` 如果运行时数据是： ```python slots = { "order_number": "10001", "order_status": "配送中", } ``` 渲染结果是： ```text 订单10001当前状态是：配送中。 ``` ##### 5.6.1.2.3 prompt `prompt` 是给 LLM 使用的指令模板。它不直接展示给用户，而是作为模型输入的一部分。在 `rephrase` 模式下，`prompt` 用于改写已经渲染好的 `text`。在 `generate` 模式下，`prompt` 用于直接生成最终回复。 `prompt` 也支持 Jinja2 模板渲染。常用参数： | 参数 | 说明 | | --- | --- | | `history` | 当前会话历史。 | | `user_message` | 当前用户最新消息。 | | `current_response` | 已渲染好的建议回复，主要用于 `rephrase`。 | 示例： ```yaml prompt: | 你是一个中文电商客服助手，语气自然、友好、简洁。请基于下面的建议回复，生成一句更自然的中文回复。对话上下文： {{ history }} 用户最后一句： {{ user_message }} 建议回复： {{ current_response }} ``` #### 5.6.1.3 Jinja2 参考 ##### 5.6.1.3.1 概述 Jinja2 是 Python 中常用的模板引擎。模板引擎用于把模板和数据合成为最终文本： ```text 模板 + 数据 -> 文本 ``` ##### 5.6.1.3.2 模板语法 ###### 5.6.1.3.2.1 变量变量输出使用 `{{ ... }}`。 ```jinja2 你好，{{ name }} ``` 传入： ```python {"name": "小明"} ``` 输出： ```text 你好，小明 ``` 如果变量值为字典或者对象，还可通过`.` 访问其中的字段，例如： ```jinja2 订单{{ slots.order_number }}正在配送中。 ``` 传入： ```python { "slots": { "order_number": "10001" } } ``` 输出： ```text 订单10001正在配送中。 ``` ###### 5.6.1.3.2.2 条件块条件块使用 `{% if ... %}` 和 `{% endif %}`。 ```jinja2 {% if history %} 对话历史： {{ history }} {% endif %} ``` 如果 `history` 有值，中间内容会输出。如果 `history` 为空，中间内容不会输出。 ##### 5.6.1.3.3 渲染工具 ###### 5.6.1.3.3.1 原生 Jinja2 使用 pip 安装： ```bash pip install jinja2 uv add jinja2 ``` 原生用法直接使用 `jinja2.Template`。 ```python from jinja2 import Template template = Template("订单{{ order_number }}正在配送中。") result = template.render(order_number="10001") ``` 渲染结果： ```text 订单10001正在配送中。 ``` ###### 5.6.1.3.3.2 LangChain PromptTemplate `langchain` 中的 `PromptTemplate` 也支持 Jinja2 模板。 ```python from langchain_core.prompts import PromptTemplate prompt = PromptTemplate.from_template( prompt_text, template_format="jinja2", ) ``` 执行时传入变量： ```python prompt_inputs = { "history": transcript, "current_response": current_response, "user_message": user_message, } chain = prompt | llm | output_parser result = await chain.ainvoke(prompt_inputs) ``` #### 5.6.1.4 代码实现整体流程：

代码实现：在action/builtin包下创建action_response.py模块。 ```python class ActionResponse(Action): # 定义一个名称 name = "action_response" async def run(self, state: DialogueState, action_kwargs: dict[str, Any]) -> ActionResult: # 获取到参数 mode = action_kwargs.get("mode", "static") if mode == "static": text = action_kwargs.get("text") render_text = self._render_text(text, state) return ActionResult( messages=[BotMessage(text=render_text)] ) elif mode == "rephrase": text = action_kwargs.get("text") render_text = self._render_text(text, state) prompt_text = action_kwargs.get("prompt") # 调用大模型 message = await self._call_llm(prompt_text, state, render_text) # 获取提示词 return ActionResult(messages=[BotMessage(text=message)]) else: prompt_text = action_kwargs.get("prompt") message = await self._call_llm(prompt_text, state) return ActionResult(messages=[BotMessage(text=message)]) def _render_text(self, text: str, state: DialogueState) -> str: template = Template(text) # 上下文，系统工作流优先于用户工作流 result = template.render(slots=state.active_task.slots if state.active_task else {}, context=state.active_system_flow or state.active_task) return result async def _call_llm(self, prompt_text: str, state: DialogueState, render_text: str = "" ) -> str: prompt = PromptTemplate.from_template(prompt_text, template_format="jinja2") chain = prompt | llm | StrOutputParser() message = await chain.ainvoke( { "history": HistoryBuilder.build(state.current_session().turns), "user_message": HistoryBuilder._render_user_message(state.pending_turn.input_message), "current_response": render_text } ) return message ``` 在atguigu/prompts包下创建一个history_builder.py模块 ```python from langchain_core.utils.mustache import render from atguigu.domain.messages import UserMessage, MessageType, BotMessage, MessageObject from atguigu.domain.state import Turn class HistoryBuilder: @staticmethod def build(turns: list[Turn]) -> str: """ 生成历史记录 :return: USER: BOT: USER: BOT: """ messages: list[str] = [] for turn in turns: # 处理用户信息 render_user_message = HistoryBuilder._render_user_message(turn.input_message) messages.append(f"USER: {render_user_message}") # 处理机器人消息 for assistant_message in turn.assistant_messages: render_bot_message = HistoryBuilder._render_bot_message(assistant_message) messages.append(f"BOT: {render_bot_message}") return "\n".join(messages) @staticmethod def _render_text(text: str) -> str: return text.strip() @staticmethod def _render_object(object: MessageObject) -> str: # [订单对象 id=A001,title="华为手机",attr1=value1,attr2=value2] title = object.title label = "订单对象" if object.type == "order" else "商品对象" id = object.id attributes = object.attributes attributes = ",".join([f"{key}={value}" for key, value in attributes.items()]) # 返回拼接字符串 return f"[{label} id = {id},title={title},{attributes}]" @staticmethod def _render_user_message(user_message: UserMessage) -> str: if user_message.type == MessageType.TEXT: return HistoryBuilder._render_text(user_message.text) else: return HistoryBuilder._render_object(user_message.object) @staticmethod def _render_bot_message(bot_message: BotMessage) -> str: if bot_message.text: return HistoryBuilder._render_text(bot_message.text) else: return HistoryBuilder._render_object(bot_message.object) ``` ### 5.6.2 action_listen `action_listen` 表示流程暂停，等待下一轮用户输入。它通常不真正生成消息，只作为 `FlowExecutor` 的停止信号。 #### 5.6.2.1 代码实现代码实现：在action/builtin包下创建action_listen.py ```python from typing import Any from atguigu.domain.state import DialogueState from atguigu.task.action.base import Action, ActionResult class ActionListen(Action): # 定义一个名称 name = "action_listen" async def run(self, state: DialogueState, action_kwargs: dict[str, Any]) -> ActionResult: pass ``` ## 5.7 自定义Action 自定义action，需要根据具体的工作流配置自行定义。 ## 5.8 ActionRunner Builder `build_action_runner()` 负责组装 `ActionRunner`。它会创建 `ActionRegistry`，再把内置 action 和自定义 action 注册进去。源码如下：在task/action包中串讲builder.py ```python def build_action_runner() -> ActionRunner: # 创建一个对象 action_runner = ActionRunner(ActionRegistry()) # 注册内置的action register_builtin_actions(action_runner) # 注册自定义的实例 register_custom_actions(action_runner) return action_runner ``` 注册Action的具体逻辑如下： ### 5.8.1 内置 Action 注册内置 action 由 `register_builtin_actions()` 注册。源码如下： ```python # 注册内置的action def register_builtin_actions(action_runner: ActionRunner): action_runner.registry.register(ActionListen()) action_runner.registry.register(ActionResponse()) ``` 这里会注册两个内置 action： | action | 说明 | | ---------------- | -------------------- | | `ActionListen` | 等待下一轮用户输入。 | | `ActionResponse` | 生成客服回复。 | ### 5.8.2 自定义 Action 注册自定义 action 由 `register_custom_actions()` 自动扫描并注册。注册逻辑如下： 1. 导入 `atguigu.task.actions.custom` 包。 2. 扫描包下面的 Python 模块。 3. 跳过子包。 4. 从模块中找出继承自 `Action` 的类。 5. 实例化 action，并注册到 `ActionRunner.registry`。 #### 5.8.2.1 技术参考自定义 Action 注册用到了 Python 的动态导入、包扫描和类检查能力。这些能力常用于插件系统、自动注册、命令发现、任务发现等场景。 ##### 5.8.2.1.1 包和模块在 Python 中，`.py` 文件通常称为模块。包含 `__init__.py` 的目录通常称为包。例如： ```text my_app/ plugins/ __init__.py send_email.py export_file.py ``` 这里： | 名称 | 含义 | | --- | --- | | `plugins` | 包。 | | `send_email.py` | 模块。 | | `export_file.py` | 模块。 | 如果要导入 `send_email.py`，完整模块名通常是： ```python my_app.plugins.send_email ``` ##### 5.8.2.1.2 动态导入普通导入是在代码里直接写： ```python import my_app.plugins.send_email ``` 动态导入是在运行时根据字符串导入模块： ```python import importlib module = importlib.import_module("my_app.plugins.send_email") ``` 动态导入适合这种场景： ```python module_name = "my_app.plugins.send_email" module = importlib.import_module(module_name) ``` 也就是说，模块名可以来自配置、扫描结果或用户输入。 ##### 5.8.2.1.3 包扫描 `pkgutil.iter_modules()` 可以扫描某个包下面有哪些模块 .py文件。示例： ```python import importlib import pkgutil package = importlib.import_module("my_app.plugins") for finder, module_name, is_pkg in pkgutil.iter_modules(package.__path__): print(finder, module_name, is_pkg) ``` `pkgutil.iter_modules()` 每次返回三个字段： | 字段 | 说明 | | --- | --- | | `finder` | 模块查找器。它知道如何在指定路径中找到并加载模块。 | | `module_name` | 模块名。如果没有传 `prefix`，通常只是短名称，例如 `send_email`。 | | `is_pkg` | 当前项是否为包。`True` 表示它是子包，`False` 表示它是普通模块。 | 例如扫描结果可能是： ```text finder=, module_name="send_email", is_pkg=False finder=, module_name="admin", is_pkg=True ``` 如果使用 `prefix`，扫描结果会直接带上完整模块名前缀： ```python for _, module_name, is_pkg in pkgutil.iter_modules( package.__path__, prefix=f"{package.__name__}.", ): print(module_name) ``` 输出可能是： ```text my_app.plugins.send_email my_app.plugins.export_file ``` 这里的 `_` 表示忽略第一个返回值 `finder`。在自动注册场景中，常见处理方式是： | 字段 | 常见用法 | | --- | --- | | `finder` | 通常不用，所以写成 `_`。 | | `module_name` | 传给 `importlib.import_module()` 动态导入模块。 | | `is_pkg` | 用来跳过子包，只处理普通 `.py` 模块。 | ##### 5.8.2.1.4 查看模块成员模块导入后，可以用 `inspect.getmembers()` 查看模块中定义了哪些对象。例如： ```python import inspect for name, obj in inspect.getmembers(module): print(name, obj) ``` `inspect.getmembers()` 每次返回两个字段： | 字段 | 说明 | | --- | --- | | `name` | 成员名称，类型是字符串。例如类名、函数名、变量名。 | | `obj` | 成员对象本身。它可能是类、函数、变量、模块等任意对象。 | 例如一个模块中有： ```python VERSION = "1.0" def hello(): pass class SendEmailPlugin: pass ``` 扫描结果中可能包含： ```text name="VERSION", obj="1.0" name="hello", obj= name="SendEmailPlugin", obj= ``` 如果只想找类，可以配合 `inspect.isclass`： ```python for name, obj in inspect.getmembers(module, inspect.isclass): print(name, obj) ``` 第二个参数 `inspect.isclass` 是一个过滤条件。只有让这个条件返回 `True` 的成员，才会出现在结果中。因此这段代码只会返回类： ```text name="SendEmailPlugin", obj= ``` 在自动注册场景中，常见处理方式是： | 字段 | 常见用法 | | --- | --- | | `name` | 通常只用于调试或日志。 | | `obj` | 用来判断是否继承指定基类，并在通过后实例化。 | ##### 5.8.2.1.5 判断继承关系 `issubclass()` 用于判断一个类是否继承自另一个类。示例： ```python class Base: pass class Child(Base): pass issubclass(Child, Base) # True ``` 扫描插件类时，常见写法是： ```python if issubclass(obj, BasePlugin) and obj is not BasePlugin: plugins.append(obj()) ``` 这里要排除 `BasePlugin` 本身，因为它只是基类，不是真正要运行的插件。 ##### 5.8.2.1.6 `__module__` 每个类都有 `__module__` 属性，用来表示这个类定义在哪个模块里。例如： ```python print(MyPlugin.__module__) ``` 如果当前模块名是： ```python my_app.plugins.send_email ``` 那么在这个模块中定义的类，通常有： ```python MyPlugin.__module__ == "my_app.plugins.send_email" ``` 扫描类时常用这个判断： ```python if obj.__module__ != module.__name__: continue ``` 它的作用是：只注册当前模块中定义的类，跳过从其他模块导入进来的类。 ##### 5.8.2.1.7 自动注册的一般流程自动注册通常可以分成五步： ```text 导入包 -> 扫描包下面的模块 -> 动态导入每个模块 -> 找出符合条件的类 -> 实例化并注册 ``` 对应到通用代码： ```python package = importlib.import_module("my_app.plugins") for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__, prefix=f"{package.__name__}."): if is_pkg: continue module = importlib.import_module(module_name) for _, obj in inspect.getmembers(module, inspect.isclass): if not issubclass(obj, BasePlugin) or obj is BasePlugin: continue # __module__：这个类是在哪个模块定义的； # 从这个模块导入的类，到底是不是这个模块中定义的！ if obj.__module__ != module.__name__: continue registry.register(obj()) ``` 这类代码的核心思想是：新增功能类时，只要把类放进约定的包里，并继承约定的基类，系统就能自动发现并注册它。最终代码： ```python import importlib import inspect import pkgutil from atguigu.task.action.base import Action from atguigu.task.action.builtin.action_listen import ActionListen from atguigu.task.action.builtin.action_response import ActionResponse from atguigu.task.action.registry import ActionRegistry from atguigu.task.action.runner import ActionRunner def register_builtin_actions(action_runner:ActionRunner): action_runner.registry.register(ActionResponse()) action_runner.registry.register(ActionListen()) def register_custom_actions(action_runner): package = importlib.import_module("atguigu.task.action.custom") for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__, prefix=f"{package.__name__}."): if is_pkg: continue module = importlib.import_module(module_name) for _, obj in inspect.getmembers(module, inspect.isclass): if not issubclass(obj, Action) or obj is Action: continue if obj.__module__ != module.__name__: continue action_runner.registry.register(obj()) def build_action_runner() -> ActionRunner: # 创建一个对象 action_runner = ActionRunner(ActionRegistry()) # 注册内置的action register_builtin_actions(action_runner) # 注册自定义的实例 register_custom_actions(action_runner) return action_runner ``` 测试： ```python if __name__ == '__main__': action_runner = build_action_runner() print(action_runner) ```

# 6. FlowExecutor ## 6.1 概述 `FlowExecutor` 负责推进当前流程。它的核心思路是： ```text 先推进 flow，直到遇到 action；再执行 action；执行完 action 后继续推进 flow；直到系统需要等待用户输入 [action_listen] ``` `FlowExecutor` 主要由两个方法协作完成： | 方法 | 作用 | | --- | --- | | `run_task()` | 外层循环，负责执行 `ActionCall` 并收集消息。 | | `advance_until_action()` | 内层循环，持续推进 flow，直到拿到下一个 `ActionCall`。 | `run_task()` 控制外层循环：

`advance_until_action()` 控制 flow 内部推进：

task执行优先级如下： ```text active_system_flow > active_task ``` ## 6.2 run_task `run_task()` 负责控制外层执行循环。源码如下：在flow包下创建executor.py模块 ```python class FlowExecutor: """ 执行流程 """ async def run_task(self, state: DialogueState, flows: FlowsList, action_runner: ActionRunner) -> list[BotMessage]: """ 用来处理一轮会话，直到遇到action_listen才会结束；否则就去执行。 :param state: :param flows: :param action_runner: :return: """ messages: list[BotMessage] = [] """ 主要推进flow :return: """ while True: action_call: ActionCall = self.advance_until_action(state, flows) # 说明循环结束了。 if action_call.action_name == "action_listen": break else: action_result: ActionResult = await action_runner.run(action_call, state) state.set_slots(action_result.slot_updates) messages.extend(action_result.messages) return messages ``` 核心逻辑： 1. 调用 `advance_until_action()` 找到下一个要执行的 action。 2. 如果 action 是 `action_listen`，说明需要等待用户输入，循环结束。 3. 否则交给 `ActionRunner` 执行 action。 4. action 执行后，写回 `slot_updates`，收集 `messages`。 5. 最终返回本轮任务轨道生成的所有 `BotMessage`。 ## 6.3 advance_until_action ### 6.3.1 概述 `advance_until_action()` 会持续推进 flow，直到遇到一个需要执行的 action。它每轮先选择当前上下文： | 状态 | 当前上下文 | | --- | --- | | 有 `active_system_flow` | 使用系统流程上下文。 | | 无系统流程，但有 `active_task` | 使用业务任务上下文。 | | 都没有 | 返回 `ActionCall("action_listen")`。 | 找到当前上下文后，它会找到当前 step，并交给 `_run_step()` 处理。 `_run_step()` 根据 step 类型分发： ```python def advance_until_action(self, state: DialogueState, flows: FlowsList) -> ActionCall: """ 推进流程，直到有action :param state:选择上下文，需要知道从哪里开始 :param flows:需要获取到flow的配置 :return: """ while True: # 1. 获取到正在执行的任务 current_task = state.current_task() if current_task is None: return ActionCall(action_name="action_listen") # 2. 根据flowId 找到flow对象 flow: Flow = flows.get_flow_by_id(current_task.flow_id) step: FlowStep = flow.get_step_by_id(current_task.step_id) # 找到具体执行的step; start 开始step不需要处理，直接走next_step； step_result: ActionCall | None = self._run_step(state, step,flows) if step_result is not None: # 说明没有找到下一步 return step_result def _run_step(self, state: DialogueState, step: FlowStep,flows: FlowsList) -> ActionCall | None: if isinstance(step, StartFlowStep): return self._run_start_step(step, state) if isinstance(step, EndFlowStep): return self._run_end_step(state) if isinstance(step, CollectSlotStep): return self._run_collect_step(step, state, flows) if isinstance(step, ActionFlowStep): return self._run_action_step(step, state) ``` state.py ```python def current_task(self): return self.active_system_task or self.active_task ``` models.py 中补充函数： ```python def get_step_by_id(self, step_id: str) -> FlowStep | None: for step in self.steps: if step.id == step_id: return step return None ``` ### 6.3.2 start step `start` step 不执行 action，只负责进入真正的业务节点。处理流程：

对应逻辑：都是在exector.py模块中 ```python def _run_start_step(self, step: StartFlowStep, state: DialogueState) -> ActionCall | None: # 寻找下一步任务 self._advance_to_next_step(step, state) return None # 封装一个获取下一个step的函数 def _advance_to_next_step(self, step: FlowStep, state: DialogueState): # 现在一下个step next_step_id = self._select_next_step(step, state) # 设置state的当前的task的step_id state.current_task().step_id = next_step_id def _select_next_step(self, step: FlowStep, state: DialogueState) -> str | None: links: list[FlowStepLink] = step.next # 判断link属性那种类型： for link in links: if isinstance(link, StaticLink): return link.target if isinstance(link, ConditionalLink): # 获取表达式 condition = link.condition data = { "slots": state.active_task.slots, "context": asdict(state.current_task()) } if bool(eval(condition, {}, data)): return link.target if isinstance(link, FallbackLink): return link.target return None ``` ### 6.3.3 collect step `collect` step 用于收集槽位。处理流程：

如果槽位缺失，会激活系统流程： ```python @dataclass class CollectSystemContext(SystemContext): slot_name: str = "" response: dict[str, Any] = field(default_factory=dict) FLOW_ID_TO_CONTEXT_CLASS={ "system_task_started": StartedSystemContext, "system_task_interrupted": InterruptedSystemContext, "system_task_canceled": CanceledSystemContext, "system_task_resumed": ResumedSystemContext, "system_collect_information": CollectSystemContext, } ``` 如果槽位已有值，会先执行校验。校验失败时会清空槽位；如果配置了失败回复，则返回 `ActionCall("action_response")`。代码实现：在executor.py ```python def _run_collect_step(self, step: CollectSlotStep, state: DialogueState, flows: FlowsList) -> ActionCall | None: self._try_to_fill_slot_from_focused_object(step, state) if state.active_task.slots.get(step.slot_name): # 有值 if step.validation: # 校验 condition = step.validation.condition if self._eval_condition(condition, state): # 校验通过 self._advance_to_next_step(step, state) return None else: # 校验失败 state.remove_slot(step.slot_name) if step.validation.failure_response: # 失败响应 return ActionCall(action_name="action_response", action_kwargs=asdict(step.validation.failure_response)) else: # 无响应 return None else: # 无需校验 self._advance_to_next_step(step, state) return None else: # 无值, 激活系统collect flow state.start_system_task(CollectSystemContext( flow_id="system_collect_information", step_id=flows.get_flow_by_id("system_collect_information").start_step().id, slot_name=step.slot_name, response=asdict(step.response) )) return None # 校验条件 def _eval_condition(self, condition: str, state: DialogueState) -> bool: data = { "slots": state.active_task.slots, "context": asdict(state.current_task()) } return bool(eval(condition, {}, data)) ``` executor.py ```python def _try_to_fill_slot_from_focused_object(self, step: CollectSlotStep, state: DialogueState): if state.focused_object is None: return if ((state.focused_object.type == "order" and step.slot_name == "order_number") or (state.focused_object.type == "product" and step.slot_name == "product_id")): state.set_slots({step.slot_name: state.focused_object.id}) ``` 在state.py 中添加 ```python def remove_slot(self, slot_name: str): self.active_task.slots.pop(slot_name) ``` 在models.py 中添加 ```python def get_flow_by_id(self, flow_id: str) -> Flow | None: for flow in self.flows: if flow.id == flow_id: return flow return None ``` ### 6.3.4 action step `action` step 会构造 `ActionCall`，交给 `run_task()` 执行。处理流程：

`ActionCall` 的结构如下： ```python ActionCall( action_name=action_name, action_kwargs=args, ) ``` `args` 只支持两种配置方式。第一种是直接配置字典： ```yaml args: mode: static text: "请告诉我你的订单号。" ``` 第二种是整体引用当前 context 中的字典属性： ```yaml args: context.response ``` 对应源码如下： ```python def _run_action_step(self, step: ActionFlowStep, state: DialogueState): # 推荐下一个step self._advance_to_next_step(step, state) # 封装一个ActionCall 对象 action_call: ActionCall = self._build_action_call(step, state) return action_call def _build_action_call(self, step: ActionFlowStep, state: DialogueState): action_name = step.action args = step.args if isinstance(args, str): args = asdict(state.current_task())[args.split(".")[1]] # 返回数据 return ActionCall(action_name=action_name, action_kwargs=args) ``` ### 6.3.5 end step `end` step 表示当前流程结束。处理流程：

处理规则如下： | 当前状态 | 处理 | | --- | --- | | 有 `active_system_flow` | 结束系统流程。 | | 有 `active_task` | 结束业务任务。 | | 都没有 | 不处理。 | 代码实现: ```python def _run_end_step(self, state: DialogueState): if state.active_system_flow: state.end_system_task() return None else: # state.active_task = None state.end_active_task() return None ``` state.py ```python def end_active_task(self): self.active_task = None ``` # 7. TaskHandler `TaskHandler` 是任务轨道的总入口。源码如下：在task包下创建handler.py ```python class TaskHandler: def __init__(self, command_processor: CommandProcessor, flows: FlowsList, flow_executor: FlowExecutor, action_runner: ActionRunner ): self.command_processor: CommandProcessor = command_processor self.flows: FlowsList = flows self.flow_executor: FlowExecutor = flow_executor self.action_runner: ActionRunner = action_runner async def handle(self, commands: list[Command], state: DialogueState) -> list[BotMessage]: self.command_processor.run(commands, state, self.flows) messages: list[BotMessage] = await self.flow_executor.run_task(state, self.flows, self.action_runner) return messages ``` 简要说明： 1. `TaskHandler` 通过构造函数接收 `CommandProcessor`、`FlowExecutor` 和 `ActionRunner`。 2. `handle()` 先确认当前轮次存在 `pending_turn`。 3. 如果有 commands，先交给 `CommandProcessor` 修改 `DialogueState`。 4. 然后调用 `FlowExecutor.run_task()` 推进 flow，并执行过程中遇到的 action。

07-电商小二-TaskHandler实现