140：产品反馈闭环与迭代：持续优化Agentic产品

class UserSurveyManager:
    def __init__(self):
        self.survey_types = {
            "nps": self.create_nps_survey,
            "csat": self.create_csat_survey,
            "feature_feedback": self.create_feature_survey,
            "general_feedback": self.create_general_survey
        }
    
    def create_survey(self, survey_type, target_users):
        """
        创建调研
        """
        if survey_type in self.survey_types:
            return self.survey_types[survey_type](target_users)
        else:
            return self.create_general_survey(target_users)
    
    def create_nps_survey(self, target_users):
        """
        创建NPS调研
        """
        return {
            "type": "nps",
            "question": "您有多大可能向朋友推荐我们的产品？",
            "scale": "0-10",
            "follow_up": "请告诉我们您给出这个分数的原因",
            "target_users": target_users,
            "created_at": datetime.now()
        }
    
    def create_csat_survey(self, target_users):
        """
        创建CSAT调研
        """
        return {
            "type": "csat",
            "question": "您对产品的整体满意度如何？",
            "scale": "1-5",
            "follow_up": "请告诉我们您的具体感受",
            "target_users": target_users,
            "created_at": datetime.now()
        }
    
    def create_feature_survey(self, target_users):
        """
        创建功能调研
        """
        return {
            "type": "feature_feedback",
            "questions": [
                "您最常使用哪些功能？",
                "哪些功能对您最有价值？",
                "您希望我们添加哪些新功能？",
                "哪些功能使用起来有困难？"
            ],
            "target_users": target_users,
            "created_at": datetime.now()
        }
    
    def distribute_survey(self, survey, distribution_channels):
        """
        分发调研
        """
        distribution_results = {}
        
        for channel in distribution_channels:
            if channel == "email":
                distribution_results[channel] = self.send_email_survey(survey)
            elif channel == "in_app":
                distribution_results[channel] = self.show_in_app_survey(survey)
            elif channel == "push":
                distribution_results[channel] = self.send_push_survey(survey)
        
        return distribution_results
    
    def collect_responses(self, survey_id):
        """
        收集响应
        """
        responses = self.get_survey_responses(survey_id)
        
        return {
            "total_responses": len(responses),
            "completion_rate": self.calculate_completion_rate(responses),
            "responses": responses
        }

class UserInterviewManager:
    def __init__(self):
        self.interview_types = {
            "exploratory": self.conduct_exploratory_interview,
            "validation": self.conduct_validation_interview,
            "feedback": self.conduct_feedback_interview
        }
    
    def schedule_interview(self, user_id, interview_type):
        """
        安排访谈
        """
        return {
            "user_id": user_id,
            "type": interview_type,
            "scheduled_time": self.find_available_slot(user_id),
            "interviewer": self.assign_interviewer(),
            "questions": self.prepare_questions(interview_type)
        }
    
    def conduct_interview(self, interview):
        """
        进行访谈
        """
        # 记录访谈内容
        transcript = self.record_interview(interview)
        
        # 分析访谈内容
        analysis = self.analyze_interview(transcript)
        
        return {
            "transcript": transcript,
            "analysis": analysis,
            "key_insights": self.extract_key_insights(analysis),
            "action_items": self.generate_action_items(analysis)
        }
    
    def prepare_questions(self, interview_type):
        """
        准备访谈问题
        """
        question_sets = {
            "exploratory": [
                "请描述您使用产品的典型场景",
                "您在使用产品时遇到的最大挑战是什么？",
                "您希望通过产品实现什么目标？",
                "您目前如何解决这个问题？"
            ],
            "validation": [
                "这个功能对您有帮助吗？",
                "您会如何使用这个功能？",
                "这个功能还缺少什么？",
                "您愿意为这个功能付费吗？"
            ],
            "feedback": [
                "您对产品的整体感受如何？",
                "哪些方面做得好？",
                "哪些方面需要改进？",
                "您有什么建议？"
            ]
        }
        
        return question_sets.get(interview_type, [])

class BehaviorDataAnalyzer:
    def __init__(self):
        self.analysis_types = {
            "usage_pattern": self.analyze_usage_pattern,
            "feature_adoption": self.analyze_feature_adoption,
            "user_flow": self.analyze_user_flow,
            "churn_prediction": self.predict_churn
        }
    
    def analyze_behavior(self, user_data, analysis_type):
        """
        分析用户行为
        """
        if analysis_type in self.analysis_types:
            return self.analysis_types[analysis_type](user_data)
        else:
            return self.analyze_usage_pattern(user_data)
    
    def analyze_usage_pattern(self, user_data):
        """
        分析使用模式
        """
        return {
            "frequency": self.calculate_usage_frequency(user_data),
            "duration": self.calculate_session_duration(user_data),
            "features_used": self.identify_used_features(user_data),
            "time_of_day": self.analyze_usage_time(user_data),
            "device_type": self.analyze_device_usage(user_data)
        }
    
    def analyze_feature_adoption(self, user_data):
        """
        分析功能采用
        """
        features = self.get_all_features()
        adoption_data = {}
        
        for feature in features:
            adoption_data[feature] = {
                "users_used": self.count_feature_users(user_data, feature),
                "usage_frequency": self.calculate_feature_frequency(user_data, feature),
                "satisfaction": self.measure_feature_satisfaction(user_data, feature)
            }
        
        return adoption_data
    
    def analyze_user_flow(self, user_data):
        """
        分析用户流程
        """
        return {
            "common_paths": self.identify_common_paths(user_data),
            "drop_off_points": self.identify_drop_off_points(user_data),
            "conversion_funnel": self.build_conversion_funnel(user_data),
            "optimization_opportunities": self.identify_optimization_opportunities(user_data)
        }
    
    def predict_churn(self, user_data):
        """
        预测流失
        """
        churn_indicators = {
            "usage_decline": self.detect_usage_decline(user_data),
            "engagement_drop": self.detect_engagement_drop(user_data),
            "support_tickets": self.analyze_support_tickets(user_data),
            "feature_abandonment": self.detect_feature_abandonment(user_data)
        }
        
        churn_score = self.calculate_churn_score(churn_indicators)
        
        return {
            "churn_score": churn_score,
            "risk_level": self.categorize_risk_level(churn_score),
            "indicators": churn_indicators,
            "recommended_actions": self.recommend_retention_actions(churn_indicators)
        }

class SupportFeedbackAnalyzer:
    def __init__(self):
        self.feedback_categories = {
            "bug_report": self.categorize_bug_report,
            "feature_request": self.categorize_feature_request,
            "usability_issue": self.categorize_usability_issue,
            "general_feedback": self.categorize_general_feedback
        }
    
    def analyze_support_tickets(self, tickets):
        """
        分析客服工单
        """
        analysis = {
            "total_tickets": len(tickets),
            "categories": {},
            "trends": {},
            "priority_issues": []
        }
        
        # 分类工单
        for ticket in tickets:
            category = self.categorize_ticket(ticket)
            if category not in analysis["categories"]:
                analysis["categories"][category] = []
            analysis["categories"][category].append(ticket)
        
        # 分析趋势
        analysis["trends"] = self.analyze_trends(analysis["categories"])
        
        # 识别优先问题
        analysis["priority_issues"] = self.identify_priority_issues(analysis["categories"])
        
        return analysis
    
    def categorize_ticket(self, ticket):
        """
        分类工单
        """
        # 使用NLP分类
        ticket_text = ticket["subject"] + " " + ticket["description"]
        
        for category, categorizer in self.feedback_categories.items():
            if categorizer(ticket_text):
                return category
        
        return "general_feedback"
    
    def categorize_bug_report(self, text):
        """
        分类Bug报告
        """
        bug_keywords = ["bug", "error", "crash", "not working", "broken", "issue"]
        return any(keyword in text.lower() for keyword in bug_keywords)
    
    def categorize_feature_request(self, text):
        """
        分类功能请求
        """
        feature_keywords = ["wish", "want", "need", "would be great", "feature", "add"]
        return any(keyword in text.lower() for keyword in feature_keywords)
    
    def categorize_usability_issue(self, text):
        """
        分类可用性问题
        """
        usability_keywords = ["confusing", "difficult", "hard to use", "unclear", "complicated"]
        return any(keyword in text.lower() for keyword in usability_keywords)
    
    def identify_priority_issues(self, categories):
        """
        识别优先问题
        """
        priority_issues = []
        
        # 基于频率和影响排序
        for category, tickets in categories.items():
            if len(tickets) > 10:  # 频繁出现的问题
                priority_issues.append({
                    "category": category,
                    "count": len(tickets),
                    "priority": "high",
                    "sample_tickets": tickets[:5]
                })
        
        return sorted(priority_issues, key=lambda x: x["count"], reverse=True)

class TrackingDataManager:
    def __init__(self):
        self.tracking_events = {}
        self.event_properties = {}
    
    def track_event(self, event_name, properties=None):
        """
        追踪事件
        """
        event_data = {
            "event_name": event_name,
            "timestamp": datetime.now(),
            "properties": properties or {},
            "user_id": self.get_current_user_id(),
            "session_id": self.get_session_id()
        }
        
        # 发送到分析平台
        self.send_to_analytics(event_data)
        
        # 存储到本地
        if event_name not in self.tracking_events:
            self.tracking_events[event_name] = []
        self.tracking_events[event_name].append(event_data)
    
    def track_page_view(self, page_name):
        """
        追踪页面浏览
        """
        self.track_event("page_view", {"page": page_name})
    
    def track_feature_usage(self, feature_name, action):
        """
        追踪功能使用
        """
        self.track_event("feature_usage", {
            "feature": feature_name,
            "action": action
        })
    
    def track_user_action(self, action_name, details=None):
        """
        追踪用户行为
        """
        self.track_event("user_action", {
            "action": action_name,
            "details": details or {}
        })
    
    def track_error(self, error_type, error_message, context=None):
        """
        追踪错误
        """
        self.track_event("error", {
            "error_type": error_type,
            "error_message": error_message,
            "context": context or {}
        })

class FeedbackStatisticsAnalyzer:
    def __init__(self):
        self.statistical_methods = {
            "descriptive": self.descriptive_statistics,
            "comparative": self.comparative_statistics,
            "trend": self.trend_analysis,
            "correlation": self.correlation_analysis
        }
    
    def analyze_statistics(self, feedback_data, method="descriptive"):
        """
        分析统计数据
        """
        if method in self.statistical_methods:
            return self.statistical_methods[method](feedback_data)
        else:
            return self.descriptive_statistics(feedback_data)
    
    def descriptive_statistics(self, data):
        """
        描述性统计
        """
        import numpy as np
        
        return {
            "count": len(data),
            "mean": np.mean(data),
            "median": np.median(data),
            "std": np.std(data),
            "min": np.min(data),
            "max": np.max(data),
            "quartiles": np.percentile(data, [25, 50, 75])
        }
    
    def comparative_statistics(self, data):
        """
        比较统计
        """
        # 比较不同群体的反馈
        segments = self.segment_data(data)
        
        comparison = {}
        for segment_name, segment_data in segments.items():
            comparison[segment_name] = self.descriptive_statistics(segment_data)
        
        return comparison
    
    def trend_analysis(self, data):
        """
        趋势分析
        """
        # 分析反馈趋势
        time_series = self.group_by_time(data)
        
        trends = {
            "overall_trend": self.calculate_overall_trend(time_series),
            "seasonality": self.detect_seasonality(time_series),
            "change_points": self.detect_change_points(time_series)
        }
        
        return trends
    
    def correlation_analysis(self, data):
        """
        相关性分析
        """
        # 分析反馈之间的相关性
        correlation_matrix = self.calculate_correlation_matrix(data)
        
        return {
            "correlation_matrix": correlation_matrix,
            "strong_correlations": self.identify_strong_correlations(correlation_matrix),
            "insights": self.generate_correlation_insights(correlation_matrix)
        }

class SentimentAnalyzer:
    def __init__(self):
        self.sentiment_model = self.load_sentiment_model()
    
    def analyze_sentiment(self, text):
        """
        分析情感
        """
        # 使用NLP模型分析情感
        sentiment_score = self.sentiment_model.predict(text)
        
        return {
            "score": sentiment_score,
            "label": self.categorize_sentiment(sentiment_score),
            "confidence": self.calculate_confidence(sentiment_score)
        }
    
    def categorize_sentiment(self, score):
        """
        分类情感
        """
        if score > 0.6:
            return "positive"
        elif score < 0.4:
            return "negative"
        else:
            return "neutral"
    
    def batch_analyze(self, texts):
        """
        批量分析
        """
        results = []
        for text in texts:
            results.append(self.analyze_sentiment(text))
        
        return {
            "results": results,
            "summary": self.summarize_sentiments(results)
        }
    
    def summarize_sentiments(self, results):
        """
        汇总情感分析结果
        """
        total = len(results)
        positive = sum(1 for r in results if r["label"] == "positive")
        negative = sum(1 for r in results if r["label"] == "negative")
        neutral = sum(1 for r in results if r["label"] == "neutral")
        
        return {
            "total": total,
            "positive": positive,
            "negative": negative,
            "neutral": neutral,
            "positive_ratio": positive / total if total > 0 else 0,
            "negative_ratio": negative / total if total > 0 else 0
        }

class TopicAnalyzer:
    def __init__(self):
        self.topic_model = self.load_topic_model()
    
    def extract_topics(self, texts):
        """
        提取主题
        """
        # 使用主题模型提取主题
        topics = self.topic_model.fit_transform(texts)
        
        return {
            "topics": self.format_topics(topics),
            "keywords": self.extract_keywords(texts),
            "topic_distribution": self.calculate_topic_distribution(topics)
        }
    
    def extract_keywords(self, texts):
        """
        提取关键词
        """
        # 使用TF-IDF或类似方法提取关键词
        keywords = []
        
        for text in texts:
            text_keywords = self.extract_text_keywords(text)
            keywords.extend(text_keywords)
        
        return self.aggregate_keywords(keywords)
    
    def categorize_feedback(self, texts):
        """
        分类反馈
        """
        categories = {
            "product_quality": [],
            "user_experience": [],
            "feature_request": [],
            "bug_report": [],
            "pricing": [],
            "support": []
        }
        
        for text in texts:
            category = self.classify_feedback(text)
            categories[category].append(text)
        
        return categories

class PriorityRanker:
    def __init__(self):
        self.priority_factors = {
            "impact": 0.4,
            "frequency": 0.3,
            "effort": 0.2,
            "strategic": 0.1
        }
    
    def rank_feedback(self, feedback_items):
        """
        排序反馈
        """
        ranked_items = []
        
        for item in feedback_items:
            score = self.calculate_priority_score(item)
            ranked_items.append({
                "item": item,
                "score": score,
                "priority": self.categorize_priority(score)
            })
        
        return sorted(ranked_items, key=lambda x: x["score"], reverse=True)
    
    def calculate_priority_score(self, item):
        """
        计算优先级得分
        """
        impact_score = self.assess_impact(item)
        frequency_score = self.assess_frequency(item)
        effort_score = self.assess_effort(item)
        strategic_score = self.assess_strategic_value(item)
        
        total_score = (
            impact_score * self.priority_factors["impact"] +
            frequency_score * self.priority_factors["frequency"] +
            (1 - effort_score) * self.priority_factors["effort"] +
            strategic_score * self.priority_factors["strategic"]
        )
        
        return total_score
    
    def assess_impact(self, item):
        """
        评估影响
        """
        # 基于用户数量、业务影响等评估
        return min(item.get("affected_users", 0) / 1000, 1.0)
    
    def assess_frequency(self, item):
        """
        评估频率
        """
        # 基于反馈出现频率评估
        return min(item.get("frequency", 0) / 100, 1.0)
    
    def assess_effort(self, item):
        """
        评估工作量
        """
        # 基于实现难度评估
        effort_levels = {"low": 0.2, "medium": 0.5, "high": 0.8}
        return effort_levels.get(item.get("effort", "medium"), 0.5)
    
    def assess_strategic_value(self, item):
        """
        评估战略价值
        """
        # 基于产品战略评估
        return item.get("strategic_value", 0.5)
    
    def categorize_priority(self, score):
        """
        分类优先级
        """
        if score >= 0.7:
            return "P0"
        elif score >= 0.5:
            return "P1"
        elif score >= 0.3:
            return "P2"
        else:
            return "P3"

class IterationPlanner:
    def __init__(self):
        self.iteration_length = 14  # 2周
        self.capacity = 40  # 每迭代40个故事点
    
    def plan_iteration(self, ranked_feedback, team_capacity):
        """
        规划迭代
        """
        iteration_plan = {
            "items": [],
            "total_points": 0,
            "capacity": team_capacity,
            "remaining_capacity": team_capacity
        }
        
        for item in ranked_feedback:
            item_points = self.estimate_story_points(item)
            
            if iteration_plan["remaining_capacity"] >= item_points:
                iteration_plan["items"].append(item)
                iteration_plan["total_points"] += item_points
                iteration_plan["remaining_capacity"] -= item_points
        
        return iteration_plan
    
    def estimate_story_points(self, item):
        """
        估算故事点
        """
        # 基于复杂度估算
        complexity = item.get("complexity", "medium")
        points_map = {"low": 1, "medium": 3, "high": 5, "very_high": 8}
        
        return points_map.get(complexity, 3)

class AgileDevelopmentManager:
    def __init__(self):
        self.sprint_length = 14
        self.current_sprint = None
    
    def start_sprint(self, sprint_plan):
        """
        开始冲刺
        """
        self.current_sprint = {
            "id": self.generate_sprint_id(),
            "start_date": datetime.now(),
            "end_date": datetime.now() + timedelta(days=self.sprint_length),
            "plan": sprint_plan,
            "status": "in_progress",
            "daily_standups": []
        }
    
    def conduct_daily_standup(self, updates):
        """
        进行每日站会
        """
        standup = {
            "date": datetime.now(),
            "updates": updates,
            "blockers": self.identify_blockers(updates)
        }
        
        self.current_sprint["daily_standups"].append(standup)
        
        return standup
    
    def complete_sprint(self):
        """
        完成冲刺
        """
        self.current_sprint["status"] = "completed"
        self.current_sprint["completion_date"] = datetime.now()
        
        # 计算完成率
        completed_items = [item for item in self.current_sprint["plan"]["items"] if item.get("completed")]
        self.current_sprint["completion_rate"] = len(completed_items) / len(self.current_sprint["plan"]["items"])
        
        return self.current_sprint

class ImprovementValidator:
    def __init__(self):
        self.test_framework = ABTestFramework()
    
    def validate_improvement(self, improvement, metrics):
        """
        验证改进效果
        """
        # 设计A/B测试
        test_design = self.design_validation_test(improvement, metrics)
        
        # 执行测试
        test_results = self.test_framework.run_test(test_design)
        
        # 分析结果
        analysis = self.analyze_test_results(test_results)
        
        return {
            "improvement": improvement,
            "test_design": test_design,
            "results": test_results,
            "analysis": analysis,
            "recommendation": self.generate_recommendation(analysis)
        }
    
    def design_validation_test(self, improvement, metrics):
        """
        设计验证测试
        """
        return {
            "name": f"validation_{improvement['id']}",
            "hypothesis": improvement["expected_impact"],
            "metrics": metrics,
            "sample_size": self.calculate_sample_size(metrics),
            "duration": 14
        }
    
    def analyze_test_results(self, results):
        """
        分析测试结果
        """
        return {
            "statistical_significance": self.check_significance(results),
            "effect_size": self.calculate_effect_size(results),
            "confidence_interval": self.calculate_confidence_interval(results),
            "practical_significance": self.check_practical_significance(results)
        }