大语言模型驱动的学生用户画像构建与个性化推荐实战指南

在数据分析领域很多初学者面临的最大挑战是如何从海量数据中提取有价值的用户洞察。传统的数据分析工具虽然功能强大但学习曲线陡峭而大语言模型的出现为数据分析带来了革命性的变化。本文将手把手教你如何利用大语言模型构建专业级的学生用户画像并实现个性化推荐即使你没有任何编程基础也能快速上手。本文特别适合教育行业从业者、数据分析初学者以及想要了解大语言模型实际应用的开发者。通过完整的实战案例你将掌握从数据准备、模型调用到结果分析的全流程并能够将这套方法应用到实际业务场景中。1. 大语言模型与数据分析的基础概念1.1 什么是大语言模型在数据分析中的应用大语言模型Large Language Model, LLM是一种基于深度学习的人工智能模型能够理解和生成人类语言。在数据分析领域LLM不再仅仅是文本生成工具而是成为了强大的数据分析助手。它能够理解自然语言查询自动执行数据清洗、特征提取、模式识别等复杂任务大大降低了数据分析的技术门槛。与传统数据分析方法相比LLM驱动的数据分析具有几个显著优势首先它允许用户使用自然语言与数据交互无需学习复杂的查询语法其次LLM具备强大的上下文理解能力能够从非结构化数据如评论、笔记中提取有价值的信息最后LLM可以自动生成分析报告和可视化建议提高分析效率。1.2 用户画像与个性化推荐的核心价值用户画像是根据用户的基本属性、行为数据、偏好信息等构建的虚拟代表它帮助系统更好地理解用户需求。在教育场景中学生用户画像可以包含学习习惯、兴趣特长、成绩表现、社交特征等多个维度。一个准确的用户画像能够为个性化教学、资源推荐、学业预警等应用提供数据支撑。个性化推荐系统则是基于用户画像为用户提供量身定制的内容或服务。在教育领域个性化推荐可以体现在课程推荐、学习资源匹配、职业规划建议等方面。有效的推荐系统不仅提高了用户体验还能显著提升学习效果和参与度。1.3 LLM如何增强传统数据分析流程传统的数据分析流程通常包括数据收集、数据清洗、特征工程、模型训练、结果分析等步骤每个环节都需要专业知识和技能。LLM的引入改变了这一现状它可以在多个环节发挥作用在数据预处理阶段LLM可以自动识别数据质量问题建议合适的清洗策略在特征工程阶段LLM能够从文本数据中提取关键特征如情感倾向、主题分类等在分析洞察阶段LLM可以生成自然语言的分析报告解释复杂的数据模式。更重要的是LLM具备强大的零样本学习能力即使在没有大量标注数据的情况下也能完成特定的分析任务这为资源有限的教育机构提供了可行的解决方案。2. 环境准备与工具选择2.1 基础环境配置开始之前我们需要准备基本的工作环境。推荐使用Python作为主要编程语言因为它拥有丰富的数据分析和AI库生态系统。以下是环境配置的基本要求Python 3.8或更高版本Jupyter Notebook或VS Code作为开发环境稳定的网络连接用于调用大语言模型API对于初学者建议使用Google Colab或类似的云端开发环境这样可以避免复杂的本地环境配置问题。这些平台通常已经预装了常用的数据分析库开箱即用。2.2 关键库安装与配置我们需要安装几个核心的Python库来支持整个数据分析流程# 数据处理和分析 pip install pandas numpy matplotlib seaborn # 大语言模型接口 pip install openai langchain # 其他工具库 pip install scikit-learn jupyter如果你使用国内网络环境可能需要配置镜像源来加速下载pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pandas numpy2.3 大语言模型API配置目前主流的大语言模型服务包括OpenAI GPT系列、百度文心一言、阿里通义千问等。本文以OpenAI API为例但基本思路适用于各种LLM服务。首先你需要注册相应的API账号并获取API密钥。然后创建一个配置文件来管理密钥# config.py import os # 设置API密钥实际使用时请从环境变量读取 os.environ[OPENAI_API_KEY] your-api-key-here os.environ[OPENAI_API_BASE] https://api.openai.com/v1重要提示在实际项目中永远不要将API密钥直接硬编码在代码中。应该使用环境变量或专门的密钥管理服务# 正确的做法从环境变量读取 import os api_key os.getenv(OPENAI_API_KEY)3. 数据准备与预处理3.1 学生数据样本设计为了构建有效的学生用户画像我们需要收集多维度数据。典型的学生数据集应包含以下信息基本属性年龄、性别、年级、专业等学业表现各科成绩、GPA、出勤率等行为数据图书馆借阅记录、在线学习时长、课程参与度兴趣特征参加的社团、课外活动、自我描述等社交互动小组合作情况、师生互动频率等下面是一个模拟的学生数据示例import pandas as pd import numpy as np # 创建示例数据集 def create_sample_student_data(num_students100): np.random.seed(42) data { student_id: [fS{str(i).zfill(3)} for i in range(1, num_students1)], age: np.random.randint(18, 25, num_students), gender: np.random.choice([男, 女], num_students), major: np.random.choice([计算机, 数学, 物理, 化学, 生物], num_students), gpa: np.round(np.random.normal(3.0, 0.5, num_students), 2), study_hours_week: np.random.randint(10, 40, num_students), library_visits_month: np.random.randint(0, 20, num_students), online_course_completion: np.random.uniform(0, 1, num_students), extracurricular_activities: np.random.randint(0, 5, num_students) } # 确保GPA在合理范围内 data[gpa] np.clip(data[gpa], 0.0, 4.0) return pd.DataFrame(data) # 生成数据 student_df create_sample_student_data() print(student_df.head())3.2 数据质量检查与清洗数据质量直接影响分析结果的可靠性。我们需要进行系统的数据质量检查def check_data_quality(df): 全面检查数据质量 print( 数据质量报告 ) print(f数据集形状: {df.shape}) print(\n缺失值统计:) print(df.isnull().sum()) print(\n数据类型:) print(df.dtypes) print(\n数值型变量描述统计:) print(df.describe()) print(\n重复值检查:) print(f重复行数: {df.duplicated().sum()}) # 执行数据质量检查 check_data_quality(student_df)常见的数据清洗操作包括处理缺失值、异常值检测和数据标准化def clean_student_data(df): 数据清洗函数 df_clean df.copy() # 处理缺失值如果有的话 if df_clean.isnull().sum().sum() 0: # 数值型变量用中位数填充 numeric_cols df_clean.select_dtypes(include[np.number]).columns df_clean[numeric_cols] df_clean[numeric_cols].fillna(df_clean[numeric_cols].median()) # 分类型变量用众数填充 categorical_cols df_clean.select_dtypes(include[object]).columns for col in categorical_cols: df_clean[col] df_clean[col].fillna(df_clean[col].mode()[0] if not df_clean[col].mode().empty else 未知) # 检测和处理异常值使用IQR方法 numeric_cols df_clean.select_dtypes(include[np.number]).columns for col in numeric_cols: Q1 df_clean[col].quantile(0.25) Q3 df_clean[col].quantile(0.75) IQR Q3 - Q1 lower_bound Q1 - 1.5 * IQR upper_bound Q3 1.5 * IQR # 将异常值限制在边界内 df_clean[col] df_clean[col].clip(lowerlower_bound, upperupper_bound) return df_clean # 执行数据清洗 student_clean clean_student_data(student_df)3.3 特征工程与数据增强为了提升画像质量我们需要从原始数据中提取更有意义的特征def create_advanced_features(df): 创建高级特征 df_enhanced df.copy() # 创建学习投入度综合指标 df_enhanced[study_engagement] ( df_enhanced[study_hours_week] / 40 # 每周学习时间占比 df_enhanced[online_course_completion] # 在线课程完成度 df_enhanced[library_visits_month] / 20 # 图书馆使用频率 ) / 3 # 创建学业表现等级 def get_academic_level(gpa): if gpa 3.5: return 优秀 elif gpa 3.0: return 良好 elif gpa 2.0: return 一般 else: return 待提升 df_enhanced[academic_level] df_enhanced[gpa].apply(get_academic_level) # 创建活动参与度分类 df_enhanced[activity_level] pd.cut( df_enhanced[extracurricular_activities], bins[-1, 0, 2, 5], labels[低参与, 中等参与, 高参与] ) return df_enhanced # 增强数据特征 student_enhanced create_advanced_features(student_clean) print(student_enhanced[[student_id, study_engagement, academic_level, activity_level]].head())4. 基于大语言模型的用户画像构建4.1 LLM驱动的特征提取策略大语言模型的核心优势在于能够从文本数据中提取深层次的特征信息。即使我们的初始数据主要是结构化数据也可以通过LLM生成丰富的文本描述进而提取更细致的用户特征。首先让我们定义一个函数使用LLM为每个学生生成综合描述from langchain.llms import OpenAI from langchain.prompts import PromptTemplate import time def generate_student_profiles(df, sample_size5): 使用LLM生成学生画像描述 # 初始化LLM使用较小的温度值保证稳定性 llm OpenAI(temperature0.3, max_tokens500) # 创建提示模板 profile_template 基于以下学生信息生成一个综合的学生画像描述重点突出学生的学习特点、潜在需求和可能的发展方向 学生ID: {student_id} 年龄: {age} 性别: {gender} 专业: {major} 平均成绩: {gpa} 每周学习时间: {study_hours}小时 图书馆月访问次数: {library_visits} 在线课程完成度: {course_completion:.1%} 课外活动数量: {activities} 学习投入度: {engagement:.2f} 学业等级: {academic_level} 活动参与度: {activity_level} 请从以下几个方面进行分析 1. 学习习惯和特点 2. 优势和待改进领域 3. 可能的兴趣发展方向 4. 个性化学习建议 分析结果 prompt PromptTemplate( input_variables[student_id, age, gender, major, gpa, study_hours, library_visits, course_completion, activities, engagement, academic_level, activity_level], templateprofile_template ) profiles [] # 为避免API限制先处理少量样本 sample_df df.head(sample_size) for _, student in sample_df.iterrows(): try: # 填充提示模板 filled_prompt prompt.format( student_idstudent[student_id], agestudent[age], genderstudent[gender], majorstudent[major], gpastudent[gpa], study_hoursstudent[study_hours_week], library_visitsstudent[library_visits_month], course_completionstudent[online_course_completion], activitiesstudent[extracurricular_activities], engagementstudent[study_engagement], academic_levelstudent[academic_level], activity_levelstudent[activity_level] ) # 调用LLM生成描述 response llm(filled_prompt) profiles.append({ student_id: student[student_id], llm_profile: response.strip() }) # 添加延迟避免API限制 time.sleep(1) except Exception as e: print(f生成学生 {student[student_id]} 画像时出错: {e}) profiles.append({ student_id: student[student_id], llm_profile: 生成失败 }) return pd.DataFrame(profiles) # 生成学生画像描述 student_profiles generate_student_profiles(student_enhanced) print(student_profiles.head())4.2 画像特征结构化提取LLM生成的文本描述虽然信息丰富但不利于后续的量化分析。我们需要将这些描述性内容转化为结构化的特征def extract_structured_traits(profiles_df): 从LLM生成的描述中提取结构化特征 llm OpenAI(temperature0.1, max_tokens300) extraction_template 请从以下学生画像描述中提取关键特征并以JSON格式返回结果。需要提取的特征包括 - learning_style: 学习风格如视觉型、听觉型、动手型等 - strengths: 主要优势列表形式 - improvement_areas: 待改进领域列表形式 - interest_categories: 兴趣类别如技术、文学、体育等 - recommendation_priority: 推荐优先级高、中、低 学生画像描述 {profile_text} 请只返回JSON格式的结果不要其他内容。 structured_traits [] for _, profile in profiles_df.iterrows(): if profile[llm_profile] ! 生成失败: try: prompt extraction_template.format(profile_textprofile[llm_profile]) response llm(prompt) # 解析JSON响应 import json traits json.loads(response.strip()) traits[student_id] profile[student_id] structured_traits.append(traits) time.sleep(1) except Exception as e: print(f解析学生 {profile[student_id]} 特征时出错: {e}) structured_traits.append({ student_id: profile[student_id], learning_style: 未知, strengths: [], improvement_areas: [], interest_categories: [], recommendation_priority: 中 }) return pd.DataFrame(structured_traits) # 提取结构化特征 structured_traits extract_structured_traits(student_profiles) print(structured_traits.head())4.3 多维度画像整合现在我们将LLM提取的特征与原始数据特征进行整合形成完整的用户画像def create_comprehensive_profiles(base_df, traits_df, profiles_df): 创建综合用户画像 # 合并所有数据 comprehensive_df base_df.merge(traits_df, onstudent_id, howleft) comprehensive_df comprehensive_df.merge(profiles_df, onstudent_id, howleft) # 创建画像摘要字段 def create_profile_summary(row): summary_parts [] summary_parts.append(f{row[gender]}性{row[age]}岁{row[major]}专业) summary_parts.append(f学业表现{row[academic_level]}GPA: {row[gpa]}) summary_parts.append(f学习投入度: {row[study_engagement]:.2f}) if pd.notna(row.get(learning_style)): summary_parts.append(f学习风格: {row[learning_style]}) if pd.notna(row.get(recommendation_priority)): summary_parts.append(f推荐优先级: {row[recommendation_priority]}) return | .join(summary_parts) comprehensive_df[profile_summary] comprehensive_df.apply(create_profile_summary, axis1) return comprehensive_df # 创建完整画像 final_profiles create_comprehensive_profiles(student_enhanced, structured_traits, student_profiles) # 查看最终结果 print( 完整用户画像示例 ) for i, (_, student) in enumerate(final_profiles.head(3).iterrows()): print(f\n学生 {student[student_id]}:) print(f摘要: {student[profile_summary]}) print(fLLM分析: {student[llm_profile][:200]}...)5. 个性化推荐系统实现5.1 推荐算法基础框架基于用户画像的推荐系统通常采用协同过滤、基于内容的推荐或混合推荐策略。我们将实现一个结合规则引擎和LLM智能推荐的混合系统class StudentRecommender: 学生个性化推荐系统 def __init__(self, profiles_df, course_catalog): self.profiles profiles_df self.course_catalog course_catalog self.llm OpenAI(temperature0.2, max_tokens400) def rule_based_recommendation(self, student_id, top_n5): 基于规则的推荐 student_data self.profiles[self.profiles[student_id] student_id].iloc[0] recommendations [] # 基于专业的推荐 major_courses self.course_catalog[ self.course_catalog[related_majors].str.contains(student_data[major], naFalse) ] # 基于学业水平的推荐 if student_data[academic_level] 优秀: difficulty_filter [进阶, 高级] elif student_data[academic_level] 良好: difficulty_filter [中级, 进阶] else: difficulty_filter [初级, 中级] level_courses self.course_catalog[ self.course_catalog[difficulty].isin(difficulty_filter) ] # 基于学习投入度的权重调整 engagement_weight student_data[study_engagement] # 合并推荐结果 all_candidates pd.concat([major_courses, level_courses]).drop_duplicates() # 简单评分算法 def calculate_score(course, student): score 0 # 专业匹配度 if student[major] in course[related_majors]: score 3 # 难度匹配度 if student[academic_level] 优秀 and course[difficulty] in [进阶, 高级]: score 2 elif student[academic_level] 良好 and course[difficulty] 中级: score 2 elif student[academic_level] in [一般, 待提升] and course[difficulty] 初级: score 2 # 学习投入度调整 score * engagement_weight return score all_candidates[score] all_candidates.apply( lambda x: calculate_score(x, student_data), axis1 ) top_recommendations all_candidates.nlargest(top_n, score) return top_recommendations[[course_id, course_name, difficulty, score]] def llm_enhanced_recommendation(self, student_id, rule_based_results): 使用LLM增强推荐结果 student_data self.profiles[self.profiles[student_id] student_id].iloc[0] recommendation_template 基于以下学生信息和初步课程推荐请生成更个性化的推荐说明 学生信息 - ID: {student_id} - 专业: {major} - 学业水平: {academic_level} - 学习风格: {learning_style} - 优势: {strengths} - 待改进领域: {improvement_areas} 初步推荐课程 {course_list} 请为每个推荐课程提供 1. 为什么这门课程适合该学生 2. 学习这门课程可能带来的好处 3. 学习建议 请以自然段落的形式返回分析结果。 # 准备课程列表文本 course_list \n.join([ f- {row[course_name]} ({row[difficulty]}难度, 匹配度: {row[score]:.2f}) for _, row in rule_based_results.iterrows() ]) prompt recommendation_template.format( student_idstudent_data[student_id], majorstudent_data[major], academic_levelstudent_data[academic_level], learning_stylestudent_data.get(learning_style, 未知), strengths, .join(student_data.get(strengths, [])), improvement_areas, .join(student_data.get(improvement_areas, [])), course_listcourse_list ) try: enhanced_analysis self.llm(prompt) return enhanced_analysis.strip() except Exception as e: return fLLM分析生成失败: {e} # 创建示例课程目录 course_catalog pd.DataFrame({ course_id: [C001, C002, C003, C004, C005], course_name: [Python编程基础, 机器学习入门, 数据结构与算法, 学术写作技巧, 时间管理 workshop], difficulty: [初级, 中级, 进阶, 初级, 初级], related_majors: [计算机, 计算机,数学, 计算机,数学, 所有专业, 所有专业] }) # 初始化推荐系统 recommender StudentRecommender(final_profiles, course_catalog) # 测试推荐功能 sample_student final_profiles.iloc[0][student_id] rule_based_recs recommender.rule_based_recommendation(sample_student) llm_enhanced recommender.llm_enhanced_recommendation(sample_student, rule_based_recs) print(基于规则的推荐结果:) print(rule_based_recs) print(\nLLM增强的推荐分析:) print(llm_enhanced)5.2 实时推荐与反馈循环一个完整的推荐系统需要能够根据用户反馈不断优化推荐结果。我们实现一个简单的反馈机制class AdaptiveRecommender(StudentRecommender): 带反馈自适应功能的推荐系统 def __init__(self, profiles_df, course_catalog): super().__init__(profiles_df, course_catalog) self.feedback_history {} def record_feedback(self, student_id, course_id, feedback_type, rating0): 记录用户反馈 if student_id not in self.feedback_history: self.feedback_history[student_id] [] feedback_record { course_id: course_id, feedback_type: feedback_type, # click, enroll, complete, rate rating: rating, timestamp: pd.Timestamp.now() } self.feedback_history[student_id].append(feedback_record) def get_adaptive_recommendations(self, student_id, top_n5): 考虑历史反馈的自适应推荐 base_recommendations self.rule_based_recommendation(student_id, top_n * 2) # 如果有反馈历史调整推荐权重 if student_id in self.feedback_history: feedback_df pd.DataFrame(self.feedback_history[student_id]) # 计算课程偏好分数 course_preferences {} for _, feedback in feedback_df.iterrows(): course_id feedback[course_id] if course_id not in course_preferences: course_preferences[course_id] 0 # 不同的反馈类型有不同的权重 weight_map {click: 0.1, enroll: 0.3, complete: 0.5, rate: feedback[rating] * 0.2} course_preferences[course_id] weight_map.get(feedback[feedback_type], 0) # 调整推荐分数 for i, course in base_recommendations.iterrows(): preference_boost course_preferences.get(course[course_id], 0) base_recommendations.at[i, score] * (1 preference_boost) # 返回调整后的top_n推荐 return base_recommendations.nlargest(top_n, score) # 测试自适应推荐 adaptive_recommender AdaptiveRecommender(final_profiles, course_catalog) # 模拟一些反馈数据 adaptive_recommender.record_feedback(sample_student, C001, click) adaptive_recommender.record_feedback(sample_student, C001, enroll) adaptive_recommender.record_feedback(sample_student, C002, click) adaptive_recs adaptive_recommender.get_adaptive_recommendations(sample_student) print(自适应推荐结果考虑用户反馈:) print(adaptive_recs)6. 系统部署与性能优化6.1 生产环境部署考虑将原型系统部署到生产环境需要考虑多个因素class ProductionRecommenderSystem: 生产环境推荐系统封装 def __init__(self, model_pathNone, api_configNone): self.model None self.api_config api_config or {} self.cache {} # 简单的缓存机制 self.request_count 0 # 加载预训练模型或配置 if model_path: self.load_model(model_path) def load_model(self, model_path): 加载预训练模型 # 这里可以加载之前训练好的模型参数 # 示例中我们使用简单的内存存储 try: # 实际项目中这里会是模型加载代码 print(f加载模型从 {model_path}) self.model_loaded True except Exception as e: print(f模型加载失败: {e}) self.model_loaded False def recommend_with_retry(self, student_id, max_retries3): 带重试机制的推荐函数 for attempt in range(max_retries): try: # 检查缓存 cache_key frec_{student_id} if cache_key in self.cache: return self.cache[cache_key] # 执行推荐逻辑 result self.generate_recommendations(student_id) # 缓存结果设置适当的过期时间 self.cache[cache_key] result self.request_count 1 return result except Exception as e: print(f推荐请求失败 (尝试 {attempt 1}/{max_retries}): {e}) if attempt max_retries - 1: return self.get_fallback_recommendations(student_id) time.sleep(2 ** attempt) # 指数退避 def generate_recommendations(self, student_id): 生成推荐的核心逻辑 # 这里集成前面实现的推荐逻辑 # 简化示例 return { student_id: student_id, recommendations: [ {course_id: C001, confidence: 0.85}, {course_id: C002, confidence: 0.78} ], timestamp: pd.Timestamp.now().isoformat() } def get_fallback_recommendations(self, student_id): 降级推荐方案 return { student_id: student_id, recommendations: [ {course_id: C000, confidence: 0.5, note: 热门基础课程} ], fallback: True, timestamp: pd.Timestamp.now().isoformat() } def get_system_metrics(self): 获取系统运行指标 return { total_requests: self.request_count, cache_size: len(self.cache), model_status: loaded if self.model_loaded else not_loaded } # 生产系统实例 production_system ProductionRecommenderSystem() metrics production_system.get_system_metrics() print(系统运行指标:, metrics)6.2 性能优化策略在大规模应用场景中性能优化至关重要import hashlib import pickle from datetime import datetime, timedelta class OptimizedRecommender(ProductionRecommenderSystem): 性能优化版的推荐系统 def __init__(self, model_pathNone, api_configNone): super().__init__(model_path, api_config) self.recommendation_cache {} self.cache_expiry timedelta(hours1) self.batch_size 10 # 批处理大小 def get_cache_key(self, student_id, additional_paramsNone): 生成缓存键 base_key frec_{student_id} if additional_params: param_str hashlib.md5(str(additional_params).encode()).hexdigest()[:8] base_key f_{param_str} return base_key def batch_recommend(self, student_ids): 批量推荐优化 results {} uncached_ids [] # 检查缓存 for student_id in student_ids: cache_key self.get_cache_key(student_id) cached_result self.recommendation_cache.get(cache_key) if cached_result and datetime.now() - cached_result[timestamp] self.cache_expiry: results[student_id] cached_result[data] else: uncached_ids.append(student_id) # 批量处理未缓存的请求 if uncached_ids: batch_results self.process_batch_recommendations(uncached_ids) results.update(batch_results) return results def process_batch_recommendations(self, student_ids): 处理批量推荐请求 batch_results {} # 分批处理避免资源过载 for i in range(0, len(student_ids), self.batch_size): batch student_ids[i:i self.batch_size] try: # 这里可以实现批量推荐逻辑 # 例如批量调用LLM API或使用向量化计算 for student_id in batch: result self.generate_recommendations(student_id) batch_results[student_id] result # 更新缓存 cache_key self.get_cache_key(student_id) self.recommendation_cache[cache_key] { data: result, timestamp: datetime.now() } # 添加延迟控制请求频率 time.sleep(0.1) except Exception as e: print(f批量处理失败: {e}) # 为失败的请求提供降级方案 for student_id in batch: if student_id not in batch_results: batch_results[student_id] self.get_fallback_recommendations(student_id) return batch_results # 测试优化版本 optimized_system OptimizedRecommender() # 模拟批量请求 test_student_ids [S001, S002, S003, S004, S005] batch_results optimized_system.batch_recommend(test_student_ids) print(批量推荐结果样本:) for student_id, result in list(batch_results.items())[:2]: print(f{student_id}: {result})7. 常见问题与解决方案7.1 API调用问题与限流处理在使用大语言模型API时经常会遇到限流和稳定性问题class RobustLLMClient: 健壮的LLM API客户端 def __init__(self, api_key, max_retries5, timeout30): self.api_key api_key self.max_retries max_retries self.timeout timeout self.last_call_time None self.min_call_interval 1.0 # 最小调用间隔秒 def call_with_backoff(self, prompt, modelgpt-3.5-turbo): 带退避机制的API调用 for attempt in range(self.max_retries): try: # 遵守速率限制 if self.last_call_time: time_since_last_call time.time() - self.last_call_time if time_since_last_call self.min_call_interval: time.sleep(self.min_call_interval - time_since_last_call) # 实际API调用代码 # 这里使用模拟响应代替真实调用 response self.simulate_api_call(prompt, model) self.last_call_time time.time() return response except Exception as e: error_msg str(e).lower() if rate limit in error_msg or 429 in error_msg: wait_time (2 ** attempt) random.uniform(0, 1) print(f速率限制等待 {wait_time:.1f} 秒后重试...) time.sleep(wait_time) elif timeout in error_msg: print(f超时错误尝试 {attempt 1}/{self.max_retries}) if attempt self.max_retries - 1: raise e else: print(fAPI调用错误: {e}) if attempt self.max_retries - 1: return self.get_fallback_response(prompt) return self.get_fallback_response(prompt) def simulate_api_call(self, prompt, model): 模拟API调用实际项目中替换为真实调用 # 模拟处理时间 time.sleep(0.5) # 模拟响应 return { choices: [ { message: { content: f这是对提示 {prompt[:50]}... 的模拟响应 } } ] } def get_fallback_response(self, prompt): 降级响应方案 return { choices: [ { message: { content: 由于系统暂时不可用返回基础推荐结果。请稍后重试。 } } ] } # 使用健壮客户端 llm_client RobustLLMClient(api_keytest-key) response llm_client.call_with_backoff(测试提示) print(API响应:, response[choices][0][message][content])