⚡ 电商小二 · 深度拆解手册

这不是速查表，是一份带你从"知道有这个东西"到"能给别人讲清楚为什么这么设计"的工程级解读。

项目一句话

用 LLM 理解用户意图 + 用代码执行具体动作，通过工程化方式驯服 LLM 的不确定性，保证业务结果稳定可控的电商智能客服系统。

三条轨道

一、请求生命周期追踪

追踪一条 {"sender_id": "u1", "text": "我要退款"} 从 HTTP 进入到回复返回的完整数据流转。这是理解整个系统的主线。

1.1 全链路鸟瞰

1.2 逐层数据快照

阶段 1：HTTP 入口 — schema 翻译

{
  "sender_id": "u1",
  "text": "我要退款"
}

FastAPI 自动将 JSON 反序列化为 ChatRequest（schema 层）。然后 Web 层手动翻译为 domain 层：

UserMessage(
    sender_id="u1",
    message_id="uuid-自动生成",
    type=MessageType.TEXT,       # 有 text → TEXT
    text="我要退款",
    object=None                   # 没带对象
)

阶段 2：Service 三步编排

# 数据库 dialogue_states 表：sender_id="u1", state_json='{"sender_id":"u1",...}'
# 反序列化后：
DialogueState(
    sender_id="u1",
    active_task=None,            # 当前没有活跃任务
    paused_tasks=[],             # 没有挂起任务
    active_system_task=None,
    focused_object=None,
    sessions=[Session(...)],     # 之前的会话历史
    current_session_id="abc-123",
    pending_turn=None
)

阶段 3：Engine 处理 — 5 个步骤

阶段 3 详细追踪：_handle_text_message

{
  "user_message": "我要退款",
  "current_conversation": "USER: 之前对话...\nBOT: 之前回复...",
  "active_task_json": null,                    # 当前无活跃任务
  "interrupted_tasks_json": "[]",              # 无挂起任务
  "focused_object_json": null,                # 无聚焦对象
  "available_flows_json": '{"flows":[{"id":"refund_request","name":"退款申请",...},...]}',
  "knowledge_intents_json": '[{"id":"product_info","description":"商品信息咨询"},...]'
}

这 7 个字段填入 Jinja2 模板 turn_plan.jinja2，通过 LangChain 链 prompt | llm | JsonOutputParser 调用 LLM：

{
  "task": {
    "commands": [
      {"command": "start_flow", "flow": "refund_request"}
    ]
  },
  "knowledge": null,
  "chitchat": null
}

TurnPlan(
    task=TaskTurnPlan(commands=[
        StartFlowCommand(command="start_flow", flow="refund_request")
    ]),
    knowledge=None,
    chitchat=None
)

第一重: commands 非空?          ✓ [StartFlowCommand(...)]
第二重: 命令类型合法?           ✓ StartFlowCommand 在 allowed 中
第三重: 不能多个 start_flow?    ✓ 只有 1 个
第四重: flow "refund_request" 存在? ✓ 在 flows 里找到
→ TurnPlanValidationResult(valid=True)

# _handle_start_flow 执行后：
state.active_task = TaskContext(
    flow_id="refund_request",
    step_id="start",            # 初始 step
    slots={}
)
state.active_system_task = StartedSystemContext(
    flow_id="system_task_started",
    started_flow_id="refund_request",
    started_flow_name="退款申请"
)

# 外层循环第 1 轮：
#   advance_until_action:
#     current_active_task = active_system_task (系统优先)
#     flow = system_task_started, step = start
#     _run_step(start) → StaticLink → step = acknowledge
#     _run_step(acknowledge) → ActionFlowStep(action="action_response")
#       → ActionCall(action_name="action_response", args={text:"好的，我们先处理退款申请。"})
#     action_name != action_listen → 执行 action
#   ActionRunner 执行 → BotMessage(text="好的，我们先处理退款申请。")
#   → acknowledge.next = end → step = end → EndFlowStep → active_system_task = None

# 外层循环第 2 轮：
#   advance_until_action:
#     current_active_task = active_task (系统过场已结束)
#     flow = refund_request, step = start
#     _run_step(start) → StaticLink → step = ask_order_number
#     ask_order_number 是 CollectFlowStep:
#       slot "order_number" 未填 → 激活 CollectSystemContext
#       → 陷入 system_collect_information 流程
#       → ActionCall(action_name="action_response", args=context.response)
#     action_name != action_listen → 执行 action
#   ActionRunner 执行 → BotMessage(text="请告诉我你的订单号。")
#   → listen step → ActionCall(action_name="action_listen") → break!

# 最终返回: [BotMessage("好的，我们先处理退款申请。"), BotMessage("请告诉我你的订单号。")]

阶段 4：收尾

DialogueState(
    sender_id="u1",
    active_task=TaskContext(flow_id="refund_request", step_id="ask_order_number", slots={}),
    paused_tasks=[],
    active_system_task=None,         # 系统过场已结束
    sessions=[..., Session(turns=[..., Turn(
        user_message=UserMessage(text="我要退款"),
        bot_messages=[
            BotMessage(text="好的，我们先处理退款申请。"),
            BotMessage(text="请告诉我你的订单号。")
        ]
    )])],
    pending_turn=None                # 已 commit
)

二、领域模型深度拆解

2.1 消息模型 (messages.py)

2.2 上下文模型 (contexts.py) — 最核心的设计

TaskContext — 用户的事

class TaskContext(BaseModel):
    flow_id: str           # 哪种表单（退款单/物流单）
    step_id: str | None    # 填到哪格了
    slots: dict = {}       # 已填好的内容

SystemContext — 系统的过场

5 个子类，每个对应一种系统过场场景：

序列化难题与解决方案

解决方案：Pydantic Discriminated Union（区分联合类型）

1. 每个子类用 Literal 固定 flow_id 值作为"身份标签"
2. 用 Annotated[Union[...], Field(discriminator="flow_id")] 组合
3. 创建 TypeAdapter 实例，用 validate_python() 反序列化

# 每个子类固定 flow_id
class StartedSystemContext(SystemContext):
    flow_id: Literal["system_task_started"] = "system_task_started"

# 联合类型 + 适配器
SystemContextUnion = Annotated[
    StartedSystemContext | InterruptedSystemContext | ...,
    Field(discriminator="flow_id")
]
system_context_adapter = TypeAdapter(SystemContextUnion)

# 反序列化时自动分发
obj = system_context_adapter.validate_python(data)
# data["flow_id"] == "system_task_started" → 自动创建 StartedSystemContext

2.3 对话状态 (state.py) — 聚合根

DialogueState 是整个系统的"中央仓库"，以 sender_id 为键，一个用户一份状态。

字段分组

pending_turn 的两步提交

current_active_task 的优先级

def current_active_task(self):
    return self.active_system_task or self.active_task

系统过场优先于业务任务——因为过场必须先说完（"好的，我们先处理退款"），然后才能让位给业务任务继续推进。

持久化：整存 JSON

一张表两列：sender_id（主键）+ state_json（TEXT）。整个 DialogueState 序列化成一个 JSON 字符串。

三、流程配置体系

3.1 YAML → Python 对象的映射

两个 YAML 文件：user_flows.yml（业务流程）和 system_flows.yml（系统流程）。由 FlowLoader 加载成 FlowsList 对象。

数据模型层次

FlowsList
├── flows: list[Flow]           # 所有流程
│   └── Flow
│       ├── id, name, description
│       ├── slots: list[FlowSlot]   # 槽位定义
│       └── steps: list[FlowStep]   # 步骤列表
│           ├── StartFlowStep
│           ├── EndFlowStep
│           ├── ActionFlowStep      # action + args
│           └── CollectFlowStep     # slot_name + response
└── slots: dict[str, FlowSlot]  # 全局槽位字典

步骤之间的连接 (links.py)

3.2 collect step 的语法糖

system_collect_information 的工作原理

# 系统流程定义
system_collect_information:
  steps:
    - id: ask
      type: action
      action: action_response
      args: context.response      # ← 关键！参数来自上下文
      next: listen
    - id: listen
      type: action
      action: action_listen
      next: end

当业务流程遇到 collect step 时，系统把 collect 的 response 放入 CollectSystemContext，然后进入 system_collect_information 流程。context.response 就是那个 response，这样同一套"回复+等待"逻辑可以复用。

3.3 条件跳转示例

- id: start
  type: start
  next:
    - if: "slots.get('product_id')"
      then: respond            # 有 product_id → 直接回复
    - else: missing_product    # 没有 → 先问

条件表达式可以读取 slots、context、flow_id、step_id。多个条件按顺序判断，命中第一个成立的分支。

四、三层架构

4.1 两套数据模型

Web 层负责两次手动翻译：ChatRequest → UserMessage（进），ProcessResult → ChatResponse（出）。FastAPI 负责自动的 JSON ↔ schema 转换。

4.2 Service 层的三步编排

async def process_message(self, user_message: UserMessage) -> ProcessResult:
    # ① 加载 (I/O)
    dialogue_state = await self.repository.load_state(user_message.sender_id)
    # ② 处理 (计算 — 纯内存，不碰数据库)
    process_result = await self.engine.process_message(dialogue_state, user_message)
    # ③ 保存 (I/O)
    await self.repository.save_state(dialogue_state)
    return process_result

4.3 Repository 层

load_state — 读取

1. 按 sender_id 查 SQL
2. 查到 → json.loads → DialogueState.model_validate
3. 没查到 → 返回全新的空 DialogueState（新用户自动给空状态，上层不用区分老/新用户）

save_state — 写入（upsert）

insert_stmt = insert(DialogueStateRecord).values(
    sender_id=dialogue_state.sender_id, state_json=state_json
)
update_stmt = insert_stmt.on_duplicate_key_update(
    state_json=insert_stmt.inserted.state_json
)
await self.session.execute(update_stmt)
await self.session.commit()

用 MySQL 的 on_duplicate_key_update——主键冲突时自动转 UPDATE。一条语句搞定插入和更新，避免"先查再决定"的两次往返和并发问题。

4.4 占位引擎模式

先写一个只回复固定话的占位 Engine，让整条链路（HTTP → Service → Repository → DB）当场跑通。下一节换真引擎时，方法签名不变，上下游一行不改。这是"面向接口编程"的好处——契约不变，实现可替换。

五、引擎核心

5.1 DialogueEngine — 调度中心

引擎自己不干活，持有一堆"专员"，只负责调度：

5.2 process_message 五步

async def process_message(self, dialogue_state, user_message) -> ProcessResult:
    # 1. 准备会话（超时判断 + 续期/新建）
    self._prepare_session(dialogue_state)
    # 2. 开启本轮 turn（begin → pending_turn）
    self._begin_turn(dialogue_state, user_message)
    # 3. 按消息类型分流
    if user_message.type is MessageType.TEXT:
        messages = await self._handle_text_message(dialogue_state)
    else:
        dialogue_state.set_focused_object(user_message.object)
        messages = await self._handle_object_message(user_message, dialogue_state)
    # 4. 回复写入 turn + commit
    dialogue_state.pending_turn.bot_messages.extend(messages)
    dialogue_state.commit_pending_turn()
    # 5. 组装返回
    return ProcessResult(sender_id=..., message_id=..., messages=messages)

_prepare_session 的三种情况

5.3 TurnPlanner — LLM 调用链

7 个 prompt 字段

三组信息：前两个让 LLM "看懂用户说什么"，中间三个让 LLM "知道现在是什么处境"，后两个给 LLM "划定能选什么"。

LangChain 链式调用

prompt_template = PromptTemplate.from_template(template=prompt_text, template_format="jinja2")
chain = prompt_template | llm | JsonOutputParser()
llm_response_dict = await chain.ainvoke(inputs_prompt)
return TurnPlan.from_dict(llm_response_dict)

5.4 Command 与 TurnPlan 数据模型

四种 Command

多态反序列化

class Command(BaseModel):
    command: str
    @classmethod
    def from_dict(cls, data):
        clz = COMMAND_NAME_TO_CLASS[data["command"]]
        return clz(**data)

COMMAND_NAME_TO_CLASS = {
    "start_flow": StartFlowCommand,
    "set_slots": SetSlotsCommand,
    "cancel_flow": CancelFlowCommand,
    "resume_flow": ResumeFlowCommand,
}

和 FlowStep 的 STEP_TYPE_TO_CLASS 一样——用映射表实现多态分发。这是项目中反复出现的设计模式。

六、防幻觉校验

6.1 为什么 LLM 输出不能直接信

6.2 校验流程 — 层层收窄

validate(state, turn_plan, flow_list, intents):
    active_tracks = _active_tracks(turn_plan)
    
    # 第一层：轨道数量
    if not active_tracks:        → MISSING_TRACK
    if len(active_tracks) > 1:   → MULTIPLE_TRACKS
    
    active_track = active_tracks[0]
    
    # 第二层：具体轨道内部
    if active_track == "task":       → _validate_task (四重校验)
    if active_track == "knowledge":  → _validate_knowledge
    # chitchat: 命中即合法，不需校验

6.3 task 四重校验

6.4 9 种 ClarifyReason 全表

6.5 ClarifyResponder 两段式设计

6.6 对象消息处理 — 三场景

七、命令处理

7.1 TaskHandler 两阶段

async def handle(self, commands, state) -> list[BotMessage]:
    # 阶段1: CommandProcessor 改状态
    self.command_processor.run(commands, state, self.flows)
    # 阶段2: FlowExecutor 推进流程，生成回复
    messages = await self.flow_executor.run_task(state, self.flows, self.action_runner)
    return messages

7.2 CommandProcessor — 逐条应用

run 遍历命令列表，_apply 按类型分发到 4 个 _handle_* 方法。顺序很重要——必须先 start_flow 创建活跃任务，set_slots 才有地方填。

7.3 _handle_start_flow 的 5 种分支

这是命令处理中最复杂的部分。同一个 start_flow 命令，根据当前 state 不同，行为完全不同：

7.4 _handle_resume_flow — 带回滚的恢复

def _handle_resume_flow(self, command, state, flows):
    # 1. 如果有活跃任务 → 先挂起（回滚到 paused_tasks）
    if state.active_task is not None:
        state.interrupted_active_task()
    
    # 2. 尝试恢复
    success = state.resumed_active_task(command.flow)
    
    if success:
        # 3a. 恢复成功 → Resumed 过场
        state.start_active_system_task(ResumedSystemContext(
            resumed_flow_id=state.active_task.flow_id,
            resumed_flow_name=flow.name
        ))
    else:
        # 3b. 没有可恢复的任务 → 走 start_flow 逻辑
        self._handle_start_flow(command, state, flows)

7.5 _handle_cancel_flow — 取消的两种场景

取消 vs 打断的本质区别：打断进 paused_tasks（可恢复），取消直接丢弃（不可恢复）。

八、Action 系统

8.1 Action 框架

Action (基类)
├── name: str                    # action 名称
├── async run(call, state) → ActionResult
│
├── ActionResponse               # 内置：回复用户
├── ActionListen                 # 内置：等待用户输入（哨兵）
│
├── ActionLookupOrderStatus      # 自定义：查订单状态
├── ActionLookupLogistics        # 自定义：查物流
└── ActionRecommendSimilar...    # 自定义：推荐相似商品

8.2 ActionResponse 的 3 种模式

变量引用

# 引用业务槽位
text: "订单{{ slots.order_number }}的退款已提交"

# 引用系统上下文
text: "好的，我们先处理{{ context.started_flow_name }}。"

# 系统流程中引用 collect 的 response
args: context.response

8.3 ActionResult

class ActionResult(BaseModel):
    slot_updates: dict = {}        # 要回写的槽位
    messages: list[BotMessage] = []  # 要回复的消息

Action 执行后可能产生两类输出：① 槽位更新（比如查到订单状态后写入 slot）；② 回复消息。FlowExecutor 会把 slot_updates 写回 state，messages 收集到最终返回列表。

8.4 ActionRegistry + ActionRunner

ActionRegistry：name → Action 实例的映射表。启动时自动扫描注册。

ActionRunner：接收 ActionCall，从 Registry 查到对应 Action，执行 run()，返回 ActionResult。

8.5 自动注册机制

# 所有 Action 子类通过装饰器或约定自动注册
class ActionLookupOrderStatus(Action):
    name = "action_lookup_order_status"
    
# build_action_runner() 时自动扫描所有 Action 子类
def build_action_runner():
    registry = ActionRegistry()
    # 自动发现并注册
    for action_cls in discover_actions():
        registry.register(action_cls())
    return ActionRunner(registry)

九、FlowExecutor — 双循环执行器

9.1 外层循环 (run_task)

async def run_task(self, state, flows, action_runner):
    messages = []
    while True:
        # 1. 找下一个要执行的 action
        action_call = self.advance_until_action(state, flows)
        
        # 2. action_listen → 等用户输入，退出
        if action_call.action_name == "action_listen":
            break
        
        # 3. 其他 action → 执行
        action_result = await action_runner.run(action_call, state)
        state.set_slots(action_result.slot_updates)
        messages.extend(action_result.messages)
    
    return messages

9.2 内层循环 (advance_until_action)

def advance_until_action(self, state, flows) -> ActionCall:
    while True:
        # 1. 获取当前任务（系统优先）
        current = state.current_active_task()
        
        # 2. 没任务 → 返回 action_listen
        if current is None:
            return ActionCall(action_name="action_listen")
        
        # 3. 获取流程 + 当前 step
        flow = flows.get_flow_by_id(current.flow_id)
        step = flow.get_step_by_id(current.step_id)
        
        # 4. 运行 step
        action_call = self._run_step(state, step, flows)
        
        # 5. step 是 action 类型 → 返回
        if action_call is not None:
            return action_call

9.3 _run_step — 4 种 step 类型的处理

9.4 完整执行示例追踪

用户说"我要退款"，CommandProcessor 已创建 active_task(refund_request, step=start) + active_system_task(Started)。

=== 外层循环第 1 轮 ===
advance_until_action:
  内层循环:
    current = active_system_task (system_task_started)
    flow = system_task_started, step = start
    _run_step(start) → 推进到 acknowledge, 返回 None
    _run_step(acknowledge) → ActionFlowStep(action_response)
      → ActionCall(action_name="action_response", args={text:"好的..."})
      → 返回 ActionCall
    → 退出内层循环
  action_name="action_response" ≠ "action_listen"
  → ActionRunner 执行 → BotMessage("好的，我们先处理退款申请。")
  → acknowledge.next = end → 推进 step 到 end
  → _run_step(end) → EndFlowStep → active_system_task = None

=== 外层循环第 2 轮 ===
advance_until_action:
  内层循环:
    current = active_task (refund_request) [系统过场已结束]
    flow = refund_request, step = start
    _run_step(start) → 推进到 ask_order_number, 返回 None
    ask_order_number 是 CollectFlowStep:
      slot "order_number" 未填
      → 激活 CollectSystemContext(slot_name="order_number", response={text:"请告诉我..."})
      → 陷入 system_collect_information 流程
      → _run_step(ask) → ActionFlowStep(action_response)
        args = context.response = {text:"请告诉我你的订单号。"}
        → ActionCall(action_name="action_response")
      → 返回 ActionCall
    → 退出内层循环
  action_name="action_response" ≠ "action_listen"
  → ActionRunner 执行 → BotMessage("请告诉我你的订单号。")
  → ask.next = listen → 推进 step 到 listen
  → _run_step(listen) → ActionFlowStep(action_listen)
    → ActionCall(action_name="action_listen")

=== 外层循环第 3 轮 ===
advance_until_action:
  内层循环:
    current = active_system_task (system_collect_information)
    flow = system_collect_information, step = listen
    _run_step(listen) → ActionFlowStep(action_listen)
      → ActionCall(action_name="action_listen")
    → 返回 ActionCall
  action_name="action_listen" == "action_listen"
  → break! 退出外层循环

→ 返回 [BotMessage("好的，我们先处理退款申请。"), BotMessage("请告诉我你的订单号。")]

9.5 条件跳转的 eval

def _select_next_step(self, step, state):
    for link in step.next:
        if isinstance(link, StaticLink):
            return link.target
        elif isinstance(link, ConditionalLink):
            # eval 条件表达式
            if eval(link.condition, {"slots": state.active_task.slots, ...}):
                return link.target
        elif isinstance(link, FallbackLink):
            fallback = link.target
    return fallback

9.6 从 focused_object 自动填槽

当流程遇到 collect step 且对应 slot 未填时，FlowExecutor 会检查 focused_object——如果聚焦对象的类型正好匹配（比如收集 order_number 且 focused_object.type == "order"），自动填入并继续推进，不需要等用户输入。

十、知识检索与闲聊

10.1 ChitChatHandler — 最轻的轨道

ChitChatHandler.handle(state):
    → ChitChatResponder.respond(user_message, recent_turns)
        → HistoryBuilder 渲染消息和历史
        → load_prompt("chitchat_respond")
        → prompt | llm | StrOutputParser()
        → [BotMessage(text=LLM回复)]

没有流程推进、没有知识检索、没有命令处理——就是把用户的话和对话历史交给 LLM，让它自然回一句。

10.2 KnowledgeHandler — 三步流程

KnowledgeHandler.handle(state, turn_plan):
    ① 从 turn_plan.knowledge.intents 获取意图列表
    ② 对每个意图：
       - 查 KnowledgeIntent → 获取 provider_ids
       - 通过 KnowledgeProviderRegistry 获取对应的 Provider
       - 调 provider.retrieve(state) → list[KnowledgeChunk]
    ③ 汇总所有 chunks → KnowledgeResponder 生成回复

10.3 KnowledgeProvider 抽象

class KnowledgeProvider(ABC):
    provider_id = ""
    
    @abstractmethod
    async def retrieve(self, state: DialogueState) -> list[KnowledgeChunk]:
        pass

class KnowledgeProviderRegistry:
    def __init__(self):
        self.providers = {}  # provider_id → Provider 实例
    
    def get(self, provider_id) -> KnowledgeProvider | None:
        return self.providers.get(provider_id)

10.4 KnowledgeIntent 注册表的三重角色

10.5 requires_object 的作用

• product_info: requires_object = "product" → 必须先有一个聚焦的商品
• order_info: requires_object = "order" → 必须先有一个聚焦的订单
• refund_policy: requires_object = None → 不需要对象，直接查 FAQ

校验时，如果意图需要对象但当前没有对应类型的 focused_object → MISSING_FOCUSED_OBJECT。

十一、设计模式总结

10 个值得记住的工程模式

多态反序列化的三种实现

十二、渐进式练习题

从理解到设计的 20 道题。建议先自己想，再展开看答案。

Level 1：理解层（能说清楚是什么）

Level 2：推理层（能推理出为什么）

Level 3：设计层（能自己设计方案）

电商小二 · 深度拆解手册 — 基于 11 篇项目笔记 + 优化文档整理

掌握一个项目的标志：你能解释每个设计决策的"为什么"，而不只是"是什么"