导读:当前AI推荐大多是"单轮查询"模式——家长问一次,AI推荐一次。但越来越多AI工具支持多轮对话,家长会逐步补充条件。本文拆解多轮对话式推荐系统的核心技术:对话状态管理、槽位填充、上下文追踪,以及教培机构如何让自己的信息适配这种新范式。
一、从单轮查询到多轮对话的演进
1.1 单轮查询的局限
传统的AI推荐流程:
用户Query → 检索 → 排序 → 推荐结果家长问:"本地有什么好的数学培训机构?"
AI返回一个推荐列表。但如果家长想要更精准的结果,只能重新组织语言再问一次。
问题:
- 家长需要把所有条件一次性说完
- AI无法利用前一轮的上下文
- 每轮都是独立的检索,没有状态延续
1.2 多轮对话的新范式
轮次1: "有什么数学机构?" → AI推荐10家 轮次2: "适合三年级的" → AI从10家中筛选出5家 轮次3: "每班不超过8人的" → AI从5家中筛选出2家 轮次4: "老师有5年以上经验的" → AI从2家中筛选出1家每一轮,家长补充一个条件,AI在上一次结果的基础上继续筛选。
这需要系统具备:
- 对话状态管理——记住所有已收集的条件
- 槽位填充——从自然语言中提取结构化字段
- 上下文追踪——知道"的"指的是上一轮的哪个结果
- 增量检索——在已有结果集上做进一步筛选
二、核心技术1:对话状态管理
2.1 状态模型定义
对话状态(Dialogue State)是所有已收集条件的集合。
from dataclasses import dataclass, field from typing import Optional, List, Dict from enum import Enum class Grade(Enum): PRESCHOOL = "preschool" G1_G3 = "g1_g3" G4_G6 = "g4_g6" JUNIOR = "junior" SENIOR = "senior" class Subject(Enum): MATH = "math" ENGLISH = "english" CHINESE = "chinese" SCIENCE = "science" @dataclass class DialogueState: """教培推荐对话状态""" subject: Optional[Subject] = None # 科目 grade: Optional[Grade] = None # 年级 max_class_size: Optional[int] = None # 最大班型 min_teacher_years: Optional[int] = None # 最低教师年限 location_radius_km: Optional[float] = None # 距离范围 budget_monthly: Optional[int] = None # 月预算上限 teaching_style: Optional[str] = None # 教学风格偏好 specific_problem: Optional[str] = None # 具体问题(如"计算粗心") # 对话元数据 turn_count: int = 0 last_updated_slot: Optional[str] = None candidate_ids: List[str] = field(default_factory=list) # 当前候选机构ID列表 def filled_slots(self) -> List[str]: """返回已填充的槽位名称""" slots = [] for f in self.__dataclass_fields__: val = getattr(self, f) if val is not None and f not in ('turn_count', 'last_updated_slot', 'candidate_ids'): slots.append(f) return slots def to_filter_dict(self) -> Dict: """转换为过滤条件字典""" return { k: v.value if isinstance(v, Enum) else v for k, v in self.__dict__.items() if v is not None and k not in ('turn_count', 'last_updated_slot', 'candidate_ids') }2.2 状态更新逻辑
每轮对话后,更新状态:
def update_state(state: DialogueState, new_slots: Dict) -> DialogueState: """根据新一轮提取的槽位更新对话状态""" for slot, value in new_slots.items(): if hasattr(state, slot): setattr(state, slot, value) state.last_updated_slot = slot state.turn_count += 1 return state # 使用示例 state = DialogueState() # 轮次1: "有什么数学机构?" state = update_state(state, {'subject': Subject.MATH}) print(f"轮次1后: {state.filled_slots()}") # ['subject'] # 轮次2: "适合三年级的" state = update_state(state, {'grade': Grade.G1_G3}) print(f"轮次2后: {state.filled_slots()}") # ['subject', 'grade'] # 轮次3: "每班不超过8人" state = update_state(state, {'max_class_size': 8}) print(f"轮次3后: {state.filled_slots()}") # ['subject', 'grade', 'max_class_size']三、核心技术2:槽位填充(Slot Filling)
3.1 基于规则的槽位填充
import re from typing import Dict, Any # 教培场景实体提取规则 SLOT_PATTERNS = { 'subject': [ (r'(数学|算数|计算|奥数)', 'math'), (r'(英语|英文|口语|雅思|托福)', 'english'), (r'(语文|作文|阅读|写作)', 'chinese'), (r'(物理|化学|生物|科学)', 'science'), ], 'grade': [ (r'(幼儿园|学前|幼小衔接)', 'preschool'), (r'(一年级|二年级|三年级|1-3年级|低年级)', 'g1_g3'), (r'(四年级|五年级|六年级|4-6年级|高年级)', 'g4_g6'), (r'(初中|中考)', 'junior'), (r'(高中|高考)', 'senior'), ], 'max_class_size': [ (r'(小班|(\d+)[人以]内|不超过(\d+)(人|个))', None), # 需要自定义解析 (r'(一对一|1对1|单独)', 1), (r'(6人班|6人以内)', 6), (r'(8人班|8人以内)', 8), (r'(10人班|10人以内)', 10), ], 'min_teacher_years': [ (r'(\d+)年(以上)?经验', None), (r'资深', 5), (r'专家', 10), ], 'specific_problem': [ (r'(计算粗心|算错|粗心)', 'calculation_careless'), (r'(注意力不集中|走神|分心)', 'attention_deficit'), (r'(基础薄弱|跟不上)', 'weak_foundation'), (r'(拔高|竞赛|提优)', 'advanced'), (r'(应用题|解决问题)', 'word_problems'), ], } def extract_slots_from_text(text: str) -> Dict[str, Any]: """从家长输入中提取槽位""" slots = {} for slot_name, patterns in SLOT_PATTERNS.items(): for pattern, fixed_value in patterns: match = re.search(pattern, text) if match: if fixed_value is not None: slots[slot_name] = fixed_value elif match.groups(): # 从捕获组中提取数字 for group in match.groups(): if group and group.isdigit(): slots[slot_name] = int(group) break break # 匹配到一个就够了 return slots # 测试 print(extract_slots_from_text("适合三年级计算粗心的小班课")) # {'grade': 'g1_g3', 'specific_problem': 'calculation_careless', 'max_class_size': '小班'} print(extract_slots_from_text("老师有5年以上经验,每班不超过8人")) # {'min_teacher_years': 5, 'max_class_size': 8}3.2 基于LLM的槽位填充
对于更复杂的场景,可以用LLM做槽位填充:
SLOT_FILLING_PROMPT = """你是一个教培推荐系统的槽位填充模块。 从家长的输入中提取以下结构化信息: 可提取的槽位: - subject: 科目(math/english/chinese/science) - grade: 年级(preschool/g1_g3/g4_g6/junior/senior) - max_class_size: 最大班型人数(整数) - min_teacher_years: 最低教师年限(整数) - location_radius_km: 距离范围(公里数) - budget_monthly: 月预算(元) - teaching_style: 教学风格偏好 - specific_problem: 具体问题 家长输入:{user_input} 当前已收集的槽位:{current_slots} 请输出JSON格式的槽位更新。只输出本轮新提取到的槽位。""" def llm_slot_filling(user_input: str, current_state: DialogueState) -> Dict: """使用LLM进行槽位填充""" prompt = SLOT_FILLING_PROMPT.format( user_input=user_input, current_slots=current_state.to_filter_dict() ) # 调用LLM API(这里用伪代码) # response = llm.chat(prompt) # return json.loads(response) pass四、核心技术3:增量检索与筛选
4.1 增量筛选流程
class IncrementalRetriever: """增量检索器:在已有候选集上做进一步筛选""" def __init__(self, institution_db): self.db = institution_db # 机构数据库 def first_round_retrieve(self, state: DialogueState, top_k: int = 20) -> List[str]: """第一轮:全库检索""" filters = state.to_filter_dict() # 使用向量检索 + 过滤条件 candidates = self.db.hybrid_search( query=self._build_query_text(state), filters=filters, top_k=top_k ) state.candidate_ids = [c.id for c in candidates] return state.candidate_ids def incremental_filter(self, state: DialogueState) -> List[str]: """后续轮次:在候选集上做增量筛选""" if not state.candidate_ids: return self.first_round_retrieve(state) new_filter = {state.last_updated_slot: getattr(state, state.last_updated_slot)} # 在已有候选集上做进一步过滤 remaining = self.db.filter_in_set( candidate_ids=state.candidate_ids, new_filter=new_filter ) # 如果筛选后太少,适当放宽 if len(remaining) < 2: relaxed = self._relax_constraints(state, remaining) state.candidate_ids = relaxed else: state.candidate_ids = remaining return state.candidate_ids def _build_query_text(self, state: DialogueState) -> str: """根据状态构建检索query""" parts = [] if state.subject: parts.append(state.subject.value) if state.grade: parts.append(state.grade.value) if state.specific_problem: parts.append(state.specific_problem) return " ".join(parts) if parts else "教培机构" def _relax_constraints(self, state: DialogueState, current: List) -> List: """当筛选结果太少时,放宽约束""" # 策略:放宽最近添加的非核心约束 non_core_slots = ['max_class_size', 'min_teacher_years', 'budget_monthly'] for slot in reversed(state.filled_slots()): if slot in non_core_slots: setattr(state, slot, None) # 重新检索 return self.db.filter_in_set( candidate_ids=state.candidate_ids, new_filter=state.to_filter_dict() ) return current五、多轮对话对教培机构信息的要求
多轮对话式推荐对机构信息有4个新要求:
5.1 信息可结构化
你的课程信息需要能被提取为结构化字段:
{ "institution": "XX教育", "courses": [ { "subject": "math", "grades": ["g1_g3", "g4_g6"], "class_size": 6, "teacher_avg_years": 8, "teaching_style": "interactive", "problems_solved": ["calculation_careless", "word_problems"] } ] }5.2 信息可增量筛选
每一轮家长补充的条件,你的信息中都要有对应字段可以被筛选。
如果你的信息只有"我们很好",任何条件都无法筛选——你就无法在多轮对话中被选中。
5.3 信息有明确边界
能说清楚自己不适合什么学生,和能说清楚自己适合什么学生一样重要。
{ "suitable_for": ["g1_g3", "calculation_careless"], "not_suitable_for": ["senior", "advanced"] }5.4 信息有优先级
在多轮对话中,核心信息优先展示。AI可能会先问科目和年级,再问班型和师资。你的信息结构也要按重要性排序。
六、完整的多轮对话推荐流程
class TutoringRecommender: """教培多轮对话推荐系统""" def __init__(self, retriever: IncrementalRetriever): self.state = DialogueState() self.retriever = retriever def process_turn(self, user_input: str) -> Dict: """处理一轮对话""" # 1. 槽位填充 new_slots = extract_slots_from_text(user_input) # 2. 更新状态 self.state = update_state(self.state, new_slots) # 3. 增量检索 candidates = self.retriever.incremental_filter(self.state) # 4. 生成推荐 results = self.retriever.db.get_details(candidates) # 5. 判断是否需要追问 follow_up = self._should_ask_follow_up() return { 'state': self.state.to_filter_dict(), 'candidates': len(results), 'results': results[:5], # 最多返回5个 'follow_up_question': follow_up, 'filled_slots': self.state.filled_slots() } def _should_ask_follow_up(self) -> Optional[str]: """判断是否需要追问家长""" if self.state.turn_count >= 5: return None if self.state.subject is None and self.state.turn_count == 0: return "请问您想给孩子找哪个科目的辅导?" if self.state.grade is None and self.state.turn_count <= 1: return "孩子现在几年级呢?" if self.state.specific_problem is None and self.state.turn_count == 2: return "孩子在学习中有什么具体的问题吗?比如计算粗心、应用题困难等?" return None # 模拟多轮对话 recommender = TutoringRecommender(retriever=IncrementalRetriever(db=None)) turns = [ "有什么数学机构", "适合三年级计算粗心的", "每班不超过8人", "老师有5年以上经验的", ] for i, turn in enumerate(turns, 1): print(f"\n=== 轮次{i} ===") print(f"家长: {turn}") result = recommender.process_turn(turn) print(f"已收集条件: {result['state']}") print(f"候选机构数: {result['candidates']}") print(f"AI追问: {result['follow_up_question']}")七、总结
| 技术模块 | 作用 | 关键技术 |
|---|---|---|
| 对话状态管理 | 记住家长所有条件 | Dataclass状态机、槽位更新 |
| 槽位填充 | 从自然语言提取结构化字段 | 正则规则/LLM抽取 |
| 增量检索 | 在候选集上做进一步筛选 | 增量过滤、约束放宽 |
| 上下文追踪 | 理解指代和省略 | 对话历史管理 |
对开发者的启示:
- 多轮对话推荐是AI推荐系统的下一个演进方向
- 对话状态管理是核心技术,需要精心设计状态模型
- 槽位填充可以混合使用规则和LLM
- 增量检索要在"精准"和"结果数量"之间平衡
对教培机构的启示:
- 信息需要可结构化——课程、师资、班型等要能被提取为字段
- 信息需要可增量筛选——每一轮条件都能对应到你的信息字段
- 信息需要有明确边界——说清楚适合和不适合什么学生