RAG系统实战:从知识库构建到检索生成的全链路设计与调优
2026/10/13 9:22:24
系列导读:教培AI推荐系统实战系列第九篇。教培机构的信息散落在官网、大众点评、知乎、小红书、微信公众号等多个平台,格式不一、口径各异、质量参差不齐。如何从这些碎片化信息中自动生成一份结构化、标准化、可直接用于AI推荐的机构画像?本文给出完整的技术方案。
一个教培机构的"数字足迹"通常包括:
| 信息源 | 典型内容 | 特点 |
|---|---|---|
| 官网 | 机构介绍、课程体系、师资团队 | 官方口径,可能夸大 |
| 大众点评 | 家长评价、评分、照片 | 真实反馈,但碎片化 |
| 知乎 | 深度分析、对比文章 | 信息密度高,但可能有偏见 |
| 小红书 | 体验分享、推荐帖 | 情感化,但可能有广告 |
| 微信公众号 | 课程宣传、活动报道 | 营销导向,时效性强 |
这些信息的共同问题是:非结构化、不统一、不可直接用于AI推荐。
我们需要一个系统,能够:
┌─────────────────────────────────────────────────────┐ │ 机构自动画像生成系统 │ ├─────────────────────────────────────────────────────┤ │ 第一层:多源信息采集 │ │ ├── 官网爬虫 │ │ ├── 点评平台API │ │ ├── 内容平台采集(知乎/小红书/公众号) │ │ └── 工商登记信息 │ │ │ │ 第二层:信息抽取与标准化 │ │ ├── EntityExtractor(实体抽取) │ │ ├── AttributeNormalizer(属性标准化) │ │ └── ConflictDetector(冲突检测) │ │ │ │ 第三层:画像融合与生成 │ │ ├── ProfileBuilder(画像构建) │ │ ├── CredibilityAssessor(可信度评估) │ │ └── QualityAuditor(质量审计) │ └─────────────────────────────────────────────────────┘import json from typing import List, Dict, Optional from dataclasses import dataclass, field @dataclass class Entity: """抽取的实体""" name: str # 实体名称 type: str # 实体类型:机构/课程/教师/地点 attributes: Dict # 属性字典 source: str # 信息来源 confidence: float # 抽取置信度 class EntityExtractor: """基于LLM的实体抽取器""" def __init__(self): # 实体类型定义 self.entity_types = { 'institution': ['name', 'type', 'location', 'founding_year', 'scale', 'certification'], 'course': ['name', 'subject', 'grade_range', 'price', 'duration', 'class_size', 'teaching_method'], 'teacher': ['name', 'title', 'education', 'experience_years', 'specialty', 'certification'], 'review': ['rating', 'sentiment', 'aspect', 'content', 'date'] } def extract(self, text: str, source: str) -> List[Entity]: """从文本中抽取实体""" # 构建LLM抽取Prompt prompt = self._build_extraction_prompt(text, source) # 调用LLM(伪代码) llm_response = self._call_llm(prompt) # 解析LLM输出 entities = self._parse_extraction(llm_response) return entities def _build_extraction_prompt(self, text: str, source: str) -> str: """构建抽取Prompt""" return f"""从以下文本中抽取教培机构相关实体。 信息来源:{source} 文本内容: {text} 请抽取以下类型的实体: 1. 机构(institution):名称、类型、位置、成立年份、规模、资质 2. 课程(course):名称、科目、年级范围、价格、课时、班型、教学方式 3. 教师(teacher):姓名、职称、学历、教龄、专长、资质 4. 评价(review):评分、情感倾向、评价维度、内容、日期 输出JSON格式,每个实体包含type、name、attributes、confidence字段。 只抽取文本中明确提到的信息,不要推测。 """ def _parse_extraction(self, llm_response: str) -> List[Entity]: """解析LLM输出""" try: data = json.loads(llm_response) entities = [] for item in data.get('entities', []): entity = Entity( name=item.get('name', ''), type=item.get('type', 'unknown'), attributes=item.get('attributes', {}), source=item.get('source', ''), confidence=item.get('confidence', 0.5) ) entities.append(entity) return entities except json.JSONDecodeError: return []@dataclass class InstitutionProfile: """机构画像""" name: str basic_info: Dict = field(default_factory=dict) # 基础信息 courses: List[Dict] = field(default_factory=list) # 课程列表 teachers: List[Dict] = field(default_factory=list) # 教师列表 reviews: List[Dict] = field(default_factory=list) # 评价汇总 reputation: Dict = field(default_factory=dict) # 口碑指标 credibility: Dict = field(default_factory=dict) # 可信度评估 metadata: Dict = field(default_factory=dict) # 元数据 class ProfileBuilder: """画像构建器""" def __init__(self): self.entity_extractor = EntityExtractor() def build(self, multi_source_data: Dict[str, str]) -> InstitutionProfile: """ 从多源数据构建画像 multi_source_data: { 'official_website': '官网文本...', 'dianping': '点评文本...', 'zhihu': '知乎文本...', 'xiaohongshu': '小红书文本...', 'wechat': '公众号文本...' } """ # 第一步:从每个数据源抽取实体 all_entities = [] for source, text in multi_source_data.items(): entities = self.entity_extractor.extract(text, source) all_entities.extend(entities) # 第二步:按类型分组 institutions = [e for e in all_entities if e.type == 'institution'] courses = [e for e in all_entities if e.type == 'course'] teachers = [e for e in all_entities if e.type == 'teacher'] reviews = [e for e in all_entities if e.type == 'review'] # 第三步:融合实体(处理同一实体的多源信息) merged_institution = self._merge_institution_entities(institutions) merged_courses = self._merge_course_entities(courses) merged_teachers = self._merge_teacher_entities(teachers) # 第四步:计算口碑指标 reputation = self._compute_reputation(reviews) # 第五步:评估可信度 credibility = self._assess_credibility(all_entities) # 构建画像 profile = InstitutionProfile( name=merged_institution.get('name', '未知机构'), basic_info=merged_institution, courses=merged_courses, teachers=merged_teachers, reviews=[r.attributes for r in reviews], reputation=reputation, credibility=credibility, metadata={ 'source_count': len(multi_source_data), 'entity_count': len(all_entities), 'build_time': self._get_timestamp() } ) return profile def _merge_institution_entities(self, entities: List[Entity]) -> Dict: """融合机构实体""" if not entities: return {} # 取置信度最高的作为基础 best = max(entities, key=lambda e: e.confidence) merged = {**best.attributes} # 从其他实体补充信息 for entity in entities: if entity is best: continue for key, value in entity.attributes.items(): if key not in merged or not merged[key]: merged[key] = value return merged def _merge_course_entities(self, entities: List[Entity]) -> List[Dict]: """融合课程实体""" # 按课程名称分组 course_groups = {} for entity in entities: name = entity.name if name not in course_groups: course_groups[name] = [] course_groups[name].append(entity) # 每组融合为一个课程 merged_courses = [] for name, group in course_groups.items(): best = max(group, key=lambda e: e.confidence) course = {'name': name, **best.attributes} # 补充其他信息 for entity in group: if entity is best: continue for key, value in entity.attributes.items(): if key not in course or not course[key]: course[key] = value merged_courses.append(course) return merged_courses def _merge_teacher_entities(self, entities: List[Entity]) -> List[Dict]: """融合教师实体""" # 按教师名称分组 teacher_groups = {} for entity in entities: name = entity.name if name not in teacher_groups: teacher_groups[name] = [] teacher_groups[name].append(entity) # 每组融合为一个教师 merged_teachers = [] for name, group in teacher_groups.items(): best = max(group, key=lambda e: e.confidence) teacher = {'name': name, **best.attributes} merged_teachers.append(teacher) return merged_teachers def _compute_reputation(self, reviews: List[Entity]) -> Dict: """计算口碑指标""" if not reviews: return {'avg_rating': 0, 'review_count': 0, 'sentiment': 'neutral'} ratings = [r.attributes.get('rating', 0) for r in reviews if r.attributes.get('rating')] sentiments = [r.attributes.get('sentiment', 'neutral') for r in reviews] avg_rating = sum(ratings) / len(ratings) if ratings else 0 # 情感分布 positive = sum(1 for s in sentiments if s == 'positive') negative = sum(1 for s in sentiments if s == 'negative') neutral = len(sentiments) - positive - negative if positive > negative * 2: overall_sentiment = 'positive' elif negative > positive * 2: overall_sentiment = 'negative' else: overall_sentiment = 'neutral' return { 'avg_rating': round(avg_rating, 2), 'review_count': len(reviews), 'sentiment': overall_sentiment, 'sentiment_distribution': { 'positive': positive, 'negative': negative, 'neutral': neutral } } def _assess_credibility(self, entities: List[Entity]) -> Dict: """评估画像可信度""" # 来源多样性 sources = set(e.source for e in entities) source_diversity = len(sources) / 5 # 假设最多5个来源 # 平均置信度 avg_confidence = sum(e.confidence for e in entities) / len(entities) if entities else 0 # 信息完整度 # ... 计算各维度的信息覆盖程度 overall_credibility = 0.4 * source_diversity + 0.4 * avg_confidence + 0.2 * 0.5 return { 'overall': round(overall_credibility, 2), 'source_diversity': round(source_diversity, 2), 'avg_confidence': round(avg_confidence, 2), 'source_count': len(sources) } def _get_timestamp(self) -> str: from datetime import datetime return datetime.now().isoformat()class QualityAuditor: """画像质量审计器""" def __init__(self): self.required_fields = { 'basic_info': ['name', 'type', 'location'], 'courses': [], # 至少有一门课程 'teachers': [], # 至少有一位教师 'reputation': ['avg_rating', 'review_count'] } def audit(self, profile: InstitutionProfile) -> Dict: """审计画像质量""" issues = [] score = 100 # 检查必填字段 for field in self.required_fields['basic_info']: if field not in profile.basic_info: issues.append(f'缺少基础信息:{field}') score -= 10 if not profile.courses: issues.append('缺少课程信息') score -= 15 if not profile.teachers: issues.append('缺少教师信息') score -= 10 # 检查信息完整度 completeness = self._check_completeness(profile) if completeness < 0.6: issues.append(f'信息完整度过低:{completeness:.1%}') score -= 10 # 检查一致性 consistency = self._check_consistency(profile) if consistency < 0.7: issues.append(f'信息一致性不足:{consistency:.1%}') score -= 10 return { 'score': max(0, score), 'level': self._score_to_level(score), 'issues': issues, 'completeness': completeness, 'consistency': consistency } def _check_completeness(self, profile: InstitutionProfile) -> float: """检查信息完整度""" # 检查各维度的信息覆盖 dimensions = { 'basic_info': len(profile.basic_info) / 6, # 6个关键字段 'courses': min(len(profile.courses) / 3, 1), # 至少3门课程 'teachers': min(len(profile.teachers) / 3, 1), # 至少3位教师 'reviews': min(profile.reputation.get('review_count', 0) / 10, 1) } return sum(dimensions.values()) / len(dimensions) def _check_consistency(self, profile: InstitutionProfile) -> float: """检查信息一致性""" # 简化版:检查是否有明显矛盾 contradictions = 0 # 检查价格区间是否合理 if profile.courses: prices = [c.get('price', 0) for c in profile.courses if c.get('price')] if prices and max(prices) > min(prices) * 10: contradictions += 1 # 检查评分是否合理 avg_rating = profile.reputation.get('avg_rating', 0) if avg_rating > 0 and (avg_rating < 1 or avg_rating > 5): contradictions += 1 # 一致性得分(0-1) return 1 - contradictions * 0.2 def _score_to_level(self, score: int) -> str: if score >= 80: return '优秀' elif score >= 60: return '良好' elif score >= 40: return '一般' else: return '较差'# 模拟多源数据 multi_source_data = { 'official_website': ''' 启航教育成立于2015年,专注少儿英语培训。 课程体系:3-6岁启蒙课程、7-12岁进阶课程。 师资团队:10位专职教师,平均教龄5年,均持有TESOL证书。 地址:市中心商业广场3楼。 ''', 'dianping': ''' 张老师(4.8分):老师很负责,孩子进步明显。 李老师(4.5分):环境不错,就是价格有点贵。 王老师(4.9分):教学方法好,孩子很喜欢。 平均评分4.7分,共156条评价。 ''', 'zhihu': ''' 启航英语在本地口碑不错,外教资源丰富。 课程体系比较完善,从启蒙到进阶都有覆盖。 价格中等偏上,一节课大概150元。 ''' } # 构建画像 builder = ProfileBuilder() profile = builder.build(multi_source_data) # 审计质量 auditor = QualityAuditor() audit_result = auditor.audit(profile) # 输出画像 print("=== 机构画像 ===") print(f"机构名称:{profile.name}") print(f"基础信息:{profile.basic_info}") print(f"课程数量:{len(profile.courses)}") print(f"教师数量:{len(profile.teachers)}") print(f"口碑评分:{profile.reputation.get('avg_rating', 0)}") print(f"可信度:{profile.credibility.get('overall', 0)}") print("\n=== 质量审计 ===") print(f"质量得分:{audit_result['score']}") print(f"质量等级:{audit_result['level']}") print(f"问题列表:{audit_result['issues']}")AI推荐系统在推荐你的机构之前,首先会"理解"你的机构——这就是画像。如果你的画像质量差(信息不全、矛盾、可信度低),AI就无法准确推荐你。
本文提出了基于LLM的教培机构自动画像生成系统: