内容纲要
任务型对话系统（详解版）
目标：精通任务型对话设计、状态管理、流程控制
目录
任务型对话概述
1. 什么是任务型对话？
【定义】
任务型对话是AI通过与用户多轮交互，收集信息、
理解意图、执行任务、完成用户目标的对话系统。
【与传统对话的区别】
维度 | 传统对话 | 任务型对话
------------|-------------------|-------------------
目标 | 开放闲聊 | 完成具体任务
状态 | 无状态 | 强状态管理
流程 | 自由对话 | 结构化流程
信息收集 | 偶然收集 | 必填槽位
执行 | 不涉及 | 任务执行
反馈 | 任意响应 | 结果验证
【核心能力】
1. 意图识别
- 理解用户想做什么
- 多意图处理
- 意图冲突解决
2. 槽位填充
- 收集必要信息
- 引导用户补充
- 默认值处理
3. 状态管理
- 跟踪对话进度
- 记忆用户信息
- 支持中断恢复
4. 任务执行
- 调用服务API
- 处理执行结果
- 错误处理与重试
5. 多轮交互
- 自然引导对话
- 上下文保持
- 确认与澄清2. 典型应用场景
【查询类】
- 订单查询
- 账户信息查询
- 余额查询
- 行程查询
【操作类】
- 下单
- 退款
- 预订
- 转账
【配置类】
- 设置提醒
- 修改个人信息
- 配置服务
- 订阅管理
【复杂任务】
- 旅游规划
- 报销申请
- 合同审批
- 投诉处理对话状态管理
1. 状态模型
from enum import Enum
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
class DialogueState(Enum):
"""对话状态"""
# 初始状态
INIT = "init"
# 意图识别中
INTENT_RECOGNITION = "intent_recognition"
# 槽位填充中
SLOT_FILLING = "slot_filling"
# 槽位确认中
SLOT_CONFIRMATION = "slot_confirmation"
# 任务执行中
TASK_EXECUTION = "task_execution"
# 结果展示中
RESULT_PRESENTATION = "result_presentation"
# 等待用户确认
AWAITING_CONFIRMATION = "awaiting_confirmation"
# 完成
COMPLETED = "completed"
# 失败
FAILED = "failed"
@dataclass
class DialogueStateInfo:
"""对话状态信息"""
session_id: str # 会话ID
user_id: str # 用户ID
# 当前状态
current_state: DialogueState
previous_state: Optional[DialogueState] = None
# 意图信息
current_intent: Optional[str] = None
intent_confidence: float = 0.0
# 槽位信息
slots: Dict[str, Slot] = field(default_factory=dict)
required_slots: List[str] = field(default_factory=list)
filled_slots: List[str] = field(default_factory=list)
pending_slots: List[str] = field(default_factory=list)
# 对话历史
dialogue_history: List[DialogueTurn] = field(default_factory=list)
# 执行结果
execution_result: Optional[Dict] = None
元数据
turn_count: int = 0
last_update: datetime = field(default_factory=datetime.now)
start_time: datetime = field(default_factory=datetime.now)2. 槽位定义
from enum import Enum
class SlotType(Enum):
"""槽位类型"""
STRING = "string" # 字符串
NUMBER = "number" # 数字
DATE = "date" # 日期
TIME = "time" # 时间
DATETIME = "datetime" # 日期时间
BOOLEAN = "boolean" # 布尔值
ENUM = "enum" # 枚举
ENTITY = "entity" # 实体
CUSTOM = "custom" # 自定义
@dataclass
class Slot:
"""槽位定义"""
name: str # 槽位名称
slot_type: SlotType # 槽位类型
# 约束条件
required: bool = True # 是否必填
multiple: bool = False # 是否多值
default_value: any = None # 默认值
# 验证规则
validation_rules: List[str] = field(default_factory=list)
allowed_values: List[str] = field(default_factory=list)
min_value: Optional[float] = None
max_value: Optional[float] = None
regex_pattern: Optional[str] = None
# 当前值
value: any = None
values: List[any] = field(default_factory=list) # 多值情况
# 状态
is_filled: bool = False
is_confirmed: bool = False
confidence: float = 0.0
# 提示信息
prompt: Optional[str] = None # 询问提示
confirmation_prompt: Optional[str] = None # 确认提示
error_prompt: Optional[str] = None # 错误提示
# 元数据
source: Optional[str] = None # 信息来源（用户/默认/推断）
extraction_time: Optional[datetime] = None
def is_valid(self) -> bool:
"""验证槽位值"""
if not self.is_filled:
return False
# 类型验证
if not self._validate_type():
return False
# 规则验证
if not self._validate_rules():
return False
return True
def _validate_type(self) -> bool:
"""类型验证"""
if self.slot_type == SlotType.NUMBER:
return isinstance(self.value, (int, float))
elif self.slot_type == SlotType.BOOLEAN:
return isinstance(self.value, bool)
elif self.slot_type == SlotType.DATE:
return hasattr(self.value, 'date')
return True
def _validate_rules(self) -> bool:
"""规则验证"""
value = self.value
# 范围验证
if self.min_value is not None and value < self.min_value:
return False
if self.max_value is not None and value > self.max_value:
return False
# 枚举验证
if self.allowed_values and value not in self.allowed_values:
return False
# 正则验证
if self.regex_pattern:
import re
if not re.match(self.regex_pattern, str(value)):
return False
return True
# 示例槽位定义
order_slots = {
"product": Slot(
name="product",
slot_type=SlotType.STRING,
required=True,
prompt="您要订购什么产品？",
confirmation_prompt="您订购的是{product}，对吗？"
),
"quantity": Slot(
name="quantity",
slot_type=SlotType.NUMBER,
required=True,
default_value=1,
min_value=1,
max_value=100,
prompt="您需要多少件？"
),
"delivery_date": Slot(
name="delivery_date",
slot_type=SlotType.DATE,
required=False,
prompt="您希望什么时候送达？（格式：YYYY-MM-DD）",
regex_pattern=r"\d{4}-\d{2}-\d{2}"
),
"payment_method": Slot(
name="payment_method",
slot_type=SlotType.ENUM,
required=True,
allowed_values=["alipay", "wechat", "bank"],
prompt="请选择支付方式：支付宝、微信、银行卡"
)
}3. 状态转换
@dataclass
class StateTransition:
"""状态转换"""
from_state: DialogueState
to_state: DialogueState
trigger: str # 触发条件
action: Optional[str] = None # 触发动作
class StateMachine:
"""状态机"""
def __init__(self, initial_state: DialogueState):
self.current_state = initial_state
self.transitions = {}
self.state_handlers = {}
def add_transition(self, transition: StateTransition):
"""添加状态转换"""
key = (transition.from_state, transition.trigger)
self.transitions[key] = transition
def add_handler(self, state: DialogueState, handler):
"""添加状态处理器"""
self.state_handlers[state] = handler
def transition(self, trigger: str, context: Dict) -> DialogueState:
"""状态转换"""
key = (self.current_state, trigger)
if key not in self.transitions:
# 无转换定义，保持当前状态
return self.current_state
transition = self.transitions[key]
# 执行转换动作
if transition.action:
self._execute_action(transition.action, context)
# 转换状态
old_state = self.current_state
self.current_state = transition.to_state
print(f"状态转换: {old_state.value} -> {transition.to_state.value}")
return self.current_state
def handle(self, context: Dict):
"""处理当前状态"""
handler = self.state_handlers.get(self.current_state)
if handler:
return handler(context)
return None
# 示例状态机定义
def create_order_state_machine() -> StateMachine:
"""创建下单流程状态机"""
sm = StateMachine(DialogueState.INIT)
# 定义状态转换
sm.add_transition(StateTransition(
from_state=DialogueState.INIT,
to_state=DialogueState.INTENT_RECOGNITION,
trigger="user_input"
))
sm.add_transition(StateTransition(
from_state=DialogueState.INTENT_RECOGNITION,
to_state=DialogueState.SLOT_FILLING,
trigger="intent_confirmed"
))
sm.add_transition(StateTransition(
from_state=DialogueState.SLOT_FILLING,
to_state=DialogueState.SLOT_CONFIRMATION,
trigger="all_slots_filled"
))
sm.add_transition(StateTransition(
from_state=DialogueState.SLOT_CONFIRMATION,
to_state=DialogueState.TASK_EXECUTION,
trigger="user_confirmed"
))
sm.add_transition(StateTransition(
from_state=DialogueState.SLOT_CONFIRMATION,
to_state=DialogueState.SLOT_FILLING,
trigger="user_corrected"
))
sm.add_transition(StateTransition(
From_state=DialogueState.TASK_EXECUTION,
to_state=DialogueState.RESULT_PRESENTATION,
trigger="execution_success"
))
sm.add_transition(StateTransition(
from_state=DialogueState.RESULT_PRESENTATION,
to_state=DialogueState.COMPLETED,
trigger="user_acknowledged"
))
# 定义状态处理器
sm.add_handler(DialogueState.INIT, lambda ctx: {
"response": "您好！请问有什么可以帮您？"
})
sm.add_handler(DialogueState.INTENT_RECOGNITION, handle_intent_recognition)
sm.add_handler(DialogueState.SLOT_FILLING, handle_slot_filling)
sm.add_handler(DialogueState.SLOT_CONFIRMATION, handle_slot_confirmation)
sm.add_handler(DialogueState.TASK_EXECUTION, handle_task_execution)
sm.add_handler(DialogueState.RESULT_PRESENTATION, handle_result_presentation)
return sm任务分解与执行
1. 任务定义
@dataclass
class Task:
"""任务定义"""
id: str # 任务ID
name: str # 任务名称
intent: str # 意图
# 槽位定义
slots: Dict[str, Slot]
# 任务描述
description: str
examples: List[str]
# 执行配置
executor: str # 执行器名称
executor_params: Dict # 执行参数
# 验证配置
validation_required: bool = True
confirmation_required: bool = True
# 错误处理
retry_count: int = 3
fallback_strategy: str = "ask_user"
# 示例任务定义
order_task = Task(
id="task_order",
name="下单",
intent="order.create",
slots=order_slots,
description="帮助用户创建订单",
examples=[
"我要订购iPhone 15",
"下单：5个苹果",
"我需要买一个MacBook Pro"
],
executor="order_service.create",
executor_params={
"api_endpoint": "/api/orders",
"timeout": 30
},
validation_required=True,
confirmation_required=True,
retry_count=2,
fallback_strategy="ask_user"
)
query_order_task = Task(
id="task_query_order",
name="查询订单",
intent="order.query",
slots={
"order_id": Slot(
name="order_id",
slot_type=SlotType.STRING,
required=True,
prompt="请提供订单号"
)
},
description="查询订单状态",
examples=[
"我的订单123456怎么样了？",
"查询订单状态：789012"
],
executor="order_service.query",
executor_params={
"api_endpoint": "/api/orders/{order_id}"
},
validation_required=False,
confirmation_required=False
)2. 任务分解
class TaskDecomizer:
"""任务分解器"""
def __init__(self, llm):
self.llm = llm
self.task_templates = {}
def register_task(self, task: Task):
"""注册任务模板"""
self.task_templates[task.intent] = task
def recognize_intent(self, user_input: str) -> Tuple[str, float]:
"""识别意图"""
# 简单关键词匹配
for intent, task in self.task_templates.items():
for example in task.examples:
similarity = self.calculate_similarity(user_input, example)
if similarity > 0.7:
return intent, similarity
# 使用LLM进行意图识别
return self._llm_intent_recognition(user_input)
def _llm_intent_recognition(self, user_input: str) -> Tuple[str, float]:
"""使用LLM识别意图"""
intents = list(self.task_templates.keys())
prompt = f"""
识别用户输入的意图。
用户输入：{user_input}
可能的意图：
{', '.join(intents)}
请返回意图名称和置信度（0-1），格式：
意图: <意图>
置信度: <数值>
"""
response = self.llm.generate(prompt)
return self.parse_intent_response(response)
def decompose_task(self, user_input: str,
context: Dict) -> Dict:
"""分解任务"""
# 1. 识别意图
intent, confidence = self.recognize_intent(user_input)
if confidence < 0.5:
return {
"status": "uncertain",
"message": "我不确定您的意图，能否再详细说明一下？"
}
# 2. 获取任务定义
task = self.task_templates.get(intent)
if not task:
return {
"status": "not_found",
"message": "抱歉，我不支持这个任务"
}
# 3. 提取槽位值
extracted_slots = self.extract_slots(user_input, task.slots)
# 4. 返回任务分解结果
return {
"status": "success",
"task": task,
"intent": intent,
"confidence": confidence,
"extracted_slots": extracted_slots,
"pending_slots": self.get_pending_slots(task, extracted_slots)
}
def extract_slots(self, user_input: str,
slots: Dict[str, Slot]) -> Dict:
"""提取槽位值"""
extracted = {}
# 使用NER提取实体
entities = self.ner_extract(user_input)
for slot_name, slot in slots.items():
# 检查实体匹配
for entity in entities:
if entity.type == slot_name:
extracted[slot_name] = {
"value": entity.value,
"confidence": entity.confidence,
"source": "extraction"
}
break
# 检查默认值
if slot.default_value and slot_name not in extracted:
extracted[slot_name] = {
"value": slot.default_value,
"confidence": 1.0,
"source": "default"
}
return extracted3. 任务执行
class TaskExecutor:
"""任务执行器"""
def __init__(self, executors: Dict[str, object]):
self.executors = executors
self.execution_history = {}
def execute(self, task: Task, slots: Dict[str, Slot]) -> Dict:
"""执行任务"""
execution_id = self.generate_execution_id()
print(f"\n=== 执行任务: {task.name} ===")
# 准备参数
params = {name: slot.value for name, slot in slots.items()}
# 获取执行器
executor = self.executors.get(task.executor)
if not executor:
return {
"status": "error",
"message": f"执行器 {task.executor} 不存在"
}
# 执行任务
try:
result = executor.execute(
params,
task.executor_params
)
self.execution_history[execution_id] = {
"task": task.id,
"params": params,
"result": result,
"timestamp": datetime.now()
}
return {
"status": "success",
"result": result,
"execution_id": execution_id
}
except Exception as e:
return {
"status": "error",
"error": str(e),
"execution_id": execution_id
}
def retry_execution(self, execution_id: str) -> Dict:
"""重试执行"""
history = self.execution_history.get(execution_id)
if not history:
return {"status": "not_found"}
# 获取原始任务
task = self.task_templates[history["task"]]
# 重新执行
return self.execute(task, history["params"])
# 示例执行器
class OrderServiceExecutor:
"""订单服务执行器"""
def __init__(self, api_client):
self.api_client = api_client
def execute(self, params: Dict, executor_params: Dict) -> Dict:
"""执行订单创建"""
endpoint = executor_params.get("api_endpoint", "/api/orders")
timeout = executor_params.get("timeout", 30)
try:
response = self.api_client.post(
endpoint,
json=params,
timeout=timeout
)
response.raise_for_status()
return {
"success": True,
"order_id": response.json()["order_id"],
"message": "订单创建成功"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": "订单创建失败"
}槽位填充与验证
1. 槽位填充策略
class SlotFillingStrategy(Enum):
"""槽位填充策略"""
DIRECT = "direct" # 直接填充
CONFIRM = "confirm" # 需要确认
GUIDE = "guide" # 引导式填充
RECOMMEND = "recommend" # 推荐式填充
class SlotFiller:
"""槽位填充器"""
def __init__(self, llm):
self.llm = llm
def fill_slot(self, slot: Slot, user_input: str,
context: Dict) -> Dict:
"""填充槽位"""
# 1. 提取值
extracted = self.extract_value(slot, user_input, context)
if not extracted:
return {
"status": "no_value",
"prompt": slot.prompt
}
# 2. 验证值
validation = self.validate_value(slot, extracted["value"])
if not validation["valid"]:
return {
"status": "invalid",
"error": validation["error"],
"prompt": slot.error_prompt or validation["error"]
}
# 3. 设置值
slot.value = extracted["value"]
slot.is_filled = True
slot.confidence = extracted.get("confidence", 1.0)
slot.source = extracted.get("source", "extraction")
slot.extraction_time = datetime.now()
return {
"status": "filled",
"slot": slot.name,
"value": slot.value
}
def extract_value(self, slot: Slot, user_input: str,
context: Dict) -> Optional[Dict]:
"""提取槽位值"""
# 根据槽位类型使用不同的提取策略
extractors = {
SlotType.DATE: self.extract_date,
SlotType.NUMBER: self.extract_number,
SlotType.STRING: self.extract_string,
SlotType.ENUM: self.extract_enum,
SlotType.BOOLEAN: self.extract_boolean
}
extractor = extractors.get(slot.slot_type, self.extract_string)
return extractor(user_input, context)
def extract_date(self, text: str, context: Dict) -> Optional[Dict]:
"""提取日期"""
# 使用日期解析库
from dateparser import parse
date_obj = parse(text, languages=['zh'])
if date_obj:
return {
"value": date_obj.date(),
"confidence": 0.9,
"source": "extraction"
}
return None
def extract_number(self, text: str, context: Dict) -> Optional[Dict]:
"""提取数字"""
import re
numbers = re.findall(r'\d+', text)
if numbers:
return {
"value": int(numbers[0]),
"confidence": 0.95,
"source": "extraction"
}
return None
def extract_enum(self, text: str, context: Dict) -> Optional[Dict]:
"""提取枚举值"""
# 使用LLM进行枚举匹配
slot = context.get("current_slot")
if not slot or not slot.allowed_values:
return None
prompt = f"""
从以下选项中选择最匹配的值：
用户输入：{text}
可选值：{', '.join(slot.allowed_values)}
只返回选中的值，不要其他内容。
"""
result = self.llm.generate(prompt).strip()
if result in slot.allowed_values:
return {
"value": result,
"confidence": 0.8,
"source": "llm_inference"
}
return None
def validate_value(self, slot: Slot, value: any) -> Dict:
"""验证槽位值"""
# 临时值验证
temp_slot = Slot(
name=slot.name,
slot_type=slot.slot_type,
value=value,
is_filled=True
)
temp_slot.validation_rules = slot.validation_rules
temp_slot.allowed_values = slot.allowed_values
temp_slot.min_value = slot.min_value
temp_slot.max_value = slot.max_value
temp_slot.regex_pattern = slot.regex_pattern
if temp_slot.is_valid():
return {"valid": True}
else:
return {
"valid": False,
"error": f"值 {value} 不符合槽位 {slot.name} 的要求"
}2. 槽位确认
class SlotConfirmator:
"""槽位确认器"""
def __init__(self, llm):
self.llm = llm
def request_confirmation(self, slots: Dict[str, Slot]) -> str:
"""请求确认"""
filled_slots = {name: slot for name, slot in slots.items()
if slot.is_filled}
# 构建确认消息
confirmation_parts = []
for name, slot in filled_slots.items():
confirmation_parts.append(f"- {name}: {slot.value}")
confirmation = "\n".join(confirmation_parts)
return f"""
请确认以下信息是否正确：
{confirmation}
请回复"确认"或"取消"，或指出需要修改的信息。
"""
def parse_confirmation(self, user_input: str,
slots: Dict[str, Slot]) -> Dict:
"""解析确认响应"""
# 检查确认
if self._is_affirmative(user_input):
return {
"status": "confirmed",
"action": "proceed"
}
# 检查取消
if self._is_negative(user_input):
return {
"status": "cancelled",
"action": "restart"
}
# 检查修正
correction = self._detect_correction(user_input, slots)
if correction:
return {
"status": "corrected",
"action": "update",
"correction": correction
}
return {
"status": "unclear",
"message": "请确认、取消或指出需要修改的信息"
}
def _is_affirmative(self, text: str) -> bool:
"""检查是否为肯定"""
affirmatives = ["确认", "是的", "对", "ok", "好的", "正确"]
return any(word in text.lower() for word in affirmatives)
def _is_negative(self, text: str) -> bool:
"""检查是否为否定"""
negatives = ["取消", "不", "错误", "重来", "不对"]
return any(word in text.lower() for word in negatives)
def _detect_correction(self, text: str,
slots: Dict[str, Slot]) -> Optional[Dict]:
"""检测修正内容"""
for slot_name, slot in slots.items():
if slot_name in text:
# 提取新值
new_value = text.replace(slot_name, "").strip()
return {
"slot": slot_name,
"new_value": new_value
}
return None多轮对话策略
1. 对话策略
class DialogueStrategy(Enum):
"""对话策略"""
QUESTION_ANSWER = "qa" # 问答式
FORM_FILLING = "form" # 表单式
CONVERSATIONAL = "conversational" # 对话式
MIXED = "mixed" # 混合式
class DialogueManager:
"""对话管理器"""
def __init__(self, task: Task, strategy: DialogueStrategy):
self.task = task
self.strategy = strategy
self.state_machine = create_order_state_machine()
self.slot_filler = SlotFiller(llm)
self.confirmator = SlotConfirmator(llm)
self.executor = TaskExecutor({})
# 对话状态
self.current_state_info = DialogueStateInfo(
session_id=self.generate_session_id(),
user_id="",
current_state=DialogueState.INITINIT,
slots={name: slot for name, slot in task.slots.items()}
)
def handle_input(self, user_input: str) -> Dict:
"""处理用户输入"""
# 记录对话轮次
turn = DialogueTurn(
user_input=user_input,
timestamp=datetime.now()
)
self.current_state_info.dialogue_history.append(turn)
self.current_state_info.turn_count += 1
# 状态机处理
context = {
"user_input": user_input,
"state_info": self.current_state_info
}
# 触发状态转换
self.state_machine.transition("user_input", context)
# 处理当前状态
response = self.state_machine.handle(context)
# 更新状态
self.current_state_info.last_update = datetime.now()
return response
# 状态处理器
def handle_intent_recognition(context: Dict) -> Dict:
"""处理意图识别"""
state_info = context["state_info"]
user_input = context["user_input"]
# 识别意图
intent, confidence = decomposer.recognize_intent(user_input)
state_info.current_intent = intent
state_info.intent_confidence = confidence
if confidence > 0.7:
# 确认意图
return {
"response": f"我理解您想要{intent_to_description(intent)}",
"action": "confirm_intent"
}
else:
# 不确定意图
return {
"response": "我不确定您的意图，您是想...",
"action": "clarify_intent"
}
def handle_slot_filling(context: Dict) -> Dict:
"""处理槽位填充"""
state_info = context["state_info"]
user_input = context["user_input"]
# 找到需要填充的槽位
pending_slot = find_next_pending_slot(state_info.slots)
if not pending_slot:
# 所有槽位已填充
return {
"response": "信息收集完成",
"action": "all_slots_filled"
}
# 填充槽位
result = slot_filler.fill_slot(pending_slot, user_input, context)
if result["status"] == "filled":
# 检查是否还有待填充槽位
next_pending = find_next_pending_slot(state_info.slots)
if next_pending:
return {
"response": f"好的，{pending_slot.name}已记录为{pending_slot.value}。{next_pending.prompt}",
"action": "continue_filling"
}
else:
return {
"response": "信息收集完成，请确认...",
"action": "all_slots_filled"
}
elif result["status"] == "invalid":
return {
"response": result["prompt"],
"action": "retry_slot"
}
else:
return {
"response": result["prompt"],
"action": "prompt_slot"
}
def handle_slot_confirmation(context: Dict) -> Dict:
"""处理槽位确认"""
state_info = context["state_info"]
user_input = context["user_input"]
# 解析确认响应
result = confirmator.parse_confirmation(
user_input,
state_info.slots
)
if result["status"] == "confirmed":
return {
"response": "好的，正在为您处理...",
"action": "execute_task"
}
elif result["status"] == "cancelled":
return {
"response": "已取消，请重新开始",
"action": "restart"
}
elif result["status"] == "corrected":
# 更新槽位
slot_name = result["correction"]["slot"]
state_info.slots[slot_name].value = result["correction"]["new_value"]
return {
"response": f"已更新{slot_name}为{result['correction']['new_value']}，还有其他需要修改的吗？",
"action": "continue_confirmation"
}
else:
return {
"response": result["message"],
"action": "reconfirm"
}
def handle_task_execution(context: Dict) -> Dict:
"""处理任务执行"""
state_info = context["state_info"]
# 执行任务
result = executor.execute(
task=order_task,
slots=state_info.slots
)
if result["status"] == "success":
state_info.execution_result = result["result"]
return {
"response": f"任务执行成功！{result['result']['message']}",
"action": "show_result"
}
else:
return {
"response": f"任务执行失败：{result['error']}",
"action": "show_error"
}
def handle_result_presentation(context: Dict) -> Dict:
"""处理结果展示"""
state_info = context["state_info"]
result = state_info.execution_result
# 格式化结果
formatted = format_result(result)
return {
"response": formatted,
"action": "completed"
}工程实现
1. 基于LangChain实现
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
class TaskOrientedChat:
"""基于LangChain的任务型对话"""
def __init__(self, task: Task):
self.task = task
self.memory = ConversationBufferMemory()
self.slots = {name: slot for name, slot in task.slots.items()}
def get_pending_slots(self) -> List[str]:
"""获取待填充槽位"""
return [
name for name, slot in self.slots.items()
if slot.required and not slot.is_filled
]
def check_completion(self) -> bool:
"""检查是否完成"""
pending = self.get_pending_slots()
return len(pending) == 0
def process(self, user_input: str) -> Dict:
"""处理用户输入"""
# 添加到记忆
self.memory.save_context(
{"input": user_input},
{"output": ""}
)
# 检查完成状态
if self.check_completion():
return self.execute_task()
# 获取待填充槽位
pending_slots = self.get_pending_slots()
if not pending_slots:
return self.request_confirmation()
# 填充槽位
next_slot = pending_slots[0]
return self.fill_slot(next_slot, user_input)
def fill_slot(self, slot_name: str, user_input: str) -> Dict:
"""填充槽位"""
slot = self.slots[slot_name]
# 使用LLM提取槽位值
prompt = f"""
从以下用户输入中提取{slot_name}的值。
用户输入：{user_input}
请只返回提取的值，不要其他内容。
"""
value = self.llm.predict(prompt).strip()
# 验证值
slot.value = value
slot.is_filled = True
# 检查是否还有待填充槽位
pending = self.get_pending_slots()
if pending:
next_slot = self.slots[pending[0]]
return {
"response": next_slot.prompt,
"status": "filling"
}
else:
return self.request_confirmation()
def request_confirmation(self) -> Dict:
"""请求确认"""
confirmation = "请确认以下信息：\n"
for name, slot in self.slots.items():
if slot.is_filled:
confirmation += f"- {name}: {slot.value}\n"
confirmation += "\n回复'确认'继续，或指出需要修改的内容。"
return {
"response": confirmation,
"status": "confirming"
}
def execute_task(self) -> Dict:
"""执行任务"""
# 准备参数
params = {name: slot.value for name, slot in self.slars.items()}
# 执行任务
result = self.executor.execute(self.task, params)
return {
"response": f"任务完成：{result['message']}",
"status": "completed",
"result": result
}实战案例
1. 机票预订对话
# 定义槽位
booking_slots = {
"departure_city": Slot(
name="departure_city",
slot_type=SlotType.STRING,
required=True,
prompt="您从哪个城市出发？",
confirmation_prompt="从{departure_city}出发"
),
"destination_city": Slot(
name="destination_city",
slot_type=SlotType.STRING,
required=True,
prompt="您要去哪个城市？"
),
"departure_date": Slot(
name="departure_date",
slot_type=SlotType.DATE,
required=True,
prompt="您希望什么时候出发？（格式：YYYY-MM-DD）",
regex_pattern=r"\d{4}-\d{2}-\d{2}"
),
"return_date": Slot(
name="return_date",
slot_type=SlotType.DATE,
required=False,
prompt="您希望什么时候返程？",
regex_pattern=r"\d{4}-\d{2}-\d{2}"
),
"passengers": Slot(
name="passengers",
slot_type=SlotType.NUMBER,
required=True,
default_value=1,
min_value=1,
max_value=9,
prompt="一共有几位乘客？"
),
"class_type": Slot(
name="class_type",
slot_type=SlotType.ENUM,
required=True,
allowed_values=["economy", "business", "first"],
default_value="economy",
prompt="请选择舱位：经济舱、商务舱、头等舱"
)
}
# 定义任务
booking_task = Task(
id="task_booking",
name="机票预订",
intent="flight.book",
slots=booking_slots,
description="帮助用户预订机票",
examples=[
"我要订机票，从北京到上海",
"预订2024-01-01从广州到深圳的机票"
],
executor="flight_service.book",
executor_params={
"api_endpoint": "/api/flights/book"
}
)
# 创建对话管理器
booking_manager = DialogueManager(
task=booking_task,
strategy=DialogueStrategy.MIXED
)
# 模拟对话
conversation = [
"我要订机票",
"从北京出发",
"去上海",
"2024年1月15日出发",
"确认",
"是的",
"好的，继续"
]
for user_input in conversation:
print(f"\n用户: {user_input}")
response = booking_manager.handle_input(user_input)
print(f"助手: {response['response']}")2. 订单查询对话
# 定义槽位
query_slots = {
"order_id": Slot(
name="order_id",
slot_type=SlotType.STRING,
required=True,
prompt="请提供您的订单号",
regex_pattern=r"[A-Z0-9]{6,20}"
)
}
# 定义任务
query_task = Task(
id="task_query",
name="订单查询",
intent="order.query",
slots=query_slots,
description="查询订单状态",
examples=[
"查询我的订单ABC123456",
"我的订单怎么样了？订单号是XYZ789012"
],
executor="order_service.query",
executor_params={
"api_endpoint": "/api/orders/{order_id}"
},
validation_required=False,
confirmation_required=False
)
# 创建对话管理器
query_manager = DialogueManager(
task=query_task,
strategy=DialogueStrategy.QUESTION_ANSWER
)
# 模拟对话
print("助手: 您好，请问有什么可以帮您？")
user_input = "查询我的订单"
print(f"\n用户: {user_input}")
response = query_manager.handle_input(user_input)
print(f"助手: {response['response']}")
user_input = "ABC123456"
print(f"\n用户: {user_input}")
response = query_manager.handle_input(user_input)
print(f"助手: {response['response']}")面试高频问法
Q1: 任务型对话的核心要素？
【标准回答】
核心要素：
1. 意图识别
- 理解用户意图
- 多意图处理
- 意图冲突解决
2. 槽位填充
- 收集必要信息
- 槽位验证
- 默认值处理
3. 状态管理
- 对话状态跟踪
- 状态转换
- 上下文维护
4. 任务执行
- 服务调用
- 结果处理
- 错误处理
5. 多轮交互
- 自然引导
- 确认机制
- 澄清能力Q2: 如何处理槽位缺失？
【标准回答】
槽位缺失处理：
1. 主动询问
- 提示用户提供
- 使用槽位prompt
2. 使用默认值
- 设置合理默认
- 减少交互轮次
3. 智能推断
- 上下文推断
- LLM推理
4. 可选标记
- 标记非必填
- 跳过缺失槽位
5. 用户提供确认
- 询问是否继续
- 提供中断选项Q3: 状态机如何设计？
【标准回答】
状态机设计：
1. 定义状态(state)
- INIT：初始状态
- INTENT_RECOGNITION：意图识别
- SLOT_FILLING：槽位填充
- TASK_EXECUTION：任务执行
- COMPLETED：完成
2. 定义事件(event)
- 用户输入
- 意图确认
- 槽位填充
- 任务完成
3. 定义转换(transition)
- state + event → next_state
- 转换条件
- 转换动作
4. 实现处理器
- 每个状态对应处理函数
- 处理业务逻辑
5. 可视化
- 状态图
- 转换路径记忆要点
【任务型对话】
明确目标
结构化流程
强状态管理
【核心要素】
意图识别：理解用户想做什么
槽位填充：收集必要信息
状态管理：跟踪对话进度
任务执行：调用服务完成
多轮交互：自然引导对话
【对话状态】
INIT：初始
INTENT_RECOGNITION：意图识别
SLOT_FILLING：槽位填充
SLOT_CONFIRMATION：槽位确认
TASK_EXECUTION：任务执行
RESULT_PRESENTATION：结果展示
COMPLETED：完成
【槽位】
类型：STRING, NUMBER, DATE, ENUM
验证：类型、范围、枚举、正则
来源：用户、默认、推断
状态：填充、确认、验证
【状态机】
状态定义
事件定义
转换规则
状态处理器文档版本: 1.0
