20.5.8 监控与效果评估（API调用、本地LLM测试）

import requests
import json
import time
from datetime import datetime

class DeepSeekMonitor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.deepseek.com/v1/chat/completions"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
    
    def query_deepseek(self, prompt, model="deepseek-chat"):
        """向DeepSeek发送查询并获取回答"""
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": "你是一个专业的搜索助手，请基于你的知识库回答用户问题。"},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,  # 低温度确保结果稳定性
            "max_tokens": 1000
        }
        
        try:
            response = requests.post(
                self.base_url, 
                headers=self.headers, 
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"API请求失败: {e}")
            return None
    
    def check_reference(self, query, target_url, target_brand):
        """检查特定URL或品牌是否在回答中被引用"""
        result = self.query_deepseek(query)
        if not result:
            return {"referenced": False, "confidence": 0}
        
        answer = result["choices"][0]["message"]["content"]
        
        # 检查URL引用
        url_referenced = target_url in answer
        
        # 检查品牌提及
        brand_referenced = target_brand.lower() in answer.lower()
        
        # 计算引用置信度（基于关键词匹配和位置）
        confidence = 0
        if url_referenced:
            confidence += 0.6
        if brand_referenced:
            confidence += 0.4
        
        return {
            "referenced": url_referenced or brand_referenced,
            "confidence": min(confidence, 1.0),
            "url_referenced": url_referenced,
            "brand_referenced": brand_referenced,
            "answer_preview": answer[:200] + "..."
        }

import csv
import json
from datetime import datetime

class DailyMonitor:
    def __init__(self, api_key, keywords_file, output_file):
        self.monitor = DeepSeekMonitor(api_key)
        self.keywords_file = keywords_file
        self.output_file = output_file
        self.results = []
    
    def load_keywords(self):
        """从CSV加载关键词和目标"""
        keywords = []
        with open(self.keywords_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                keywords.append(row)
        return keywords
    
    def run_daily_check(self):
        """执行每日监控检查"""
        keywords = self.load_keywords()
        today = datetime.now().strftime("%Y-%m-%d")
        
        for kw in keywords:
            print(f"检查关键词: {kw['query']}")
            result = self.monitor.check_reference(
                kw['query'],
                kw['target_url'],
                kw['target_brand']
            )
            
            result['keyword'] = kw['query']
            result['date'] = today
            result['category'] = kw.get('category', 'general')
            
            self.results.append(result)
            
            # 避免API限流
            time.sleep(1)
        
        self.save_results()
    
    def save_results(self):
        """保存结果到JSON文件"""
        with open(self.output_file, 'a', encoding='utf-8') as f:
            for result in self.results:
                f.write(json.dumps(result, ensure_ascii=False) + '\n')
        
        print(f"结果已保存到 {self.output_file}")

import json
from collections import defaultdict
from datetime import datetime, timedelta

class ReferenceAnalyzer:
    def __init__(self, data_file):
        self.data_file = data_file
        self.data = self.load_data()
    
    def load_data(self):
        """加载监控数据"""
        data = []
        with open(self.data_file, 'r', encoding='utf-8') as f:
            for line in f:
                if line.strip():
                    data.append(json.loads(line))
        return data
    
    def calculate_reference_rate(self, days=7):
        """计算指定天数的引用率"""
        cutoff = datetime.now() - timedelta(days=days)
        
        # 按关键词分组
        keyword_stats = defaultdict(lambda: {"total": 0, "referenced": 0})
        
        for record in self.data:
            record_date = datetime.strptime(record['date'], "%Y-%m-%d")
            if record_date >= cutoff:
                kw = record['keyword']
                keyword_stats[kw]["total"] += 1
                if record['referenced']:
                    keyword_stats[kw]["referenced"] += 1
        
        # 计算引用率
        results = {}
        for kw, stats in keyword_stats.items():
            rate = stats["referenced"] / stats["total"] * 100 if stats["total"] > 0 else 0
            results[kw] = {
                "total_checks": stats["total"],
                "referenced_count": stats["referenced"],
                "reference_rate": round(rate, 2)
            }
        
        return results
    
    def trend_analysis(self, keyword):
        """分析特定关键词的引用趋势"""
        trend_data = []
        for record in self.data:
            if record['keyword'] == keyword:
                trend_data.append({
                    "date": record['date'],
                    "referenced": record['referenced'],
                    "confidence": record['confidence']
                })
        
        # 按日期排序
        trend_data.sort(key=lambda x: x['date'])
        return trend_data

import requests
import json

class LocalDeepSeekTester:
    def __init__(self, base_url="http://localhost:11434"):
        self.base_url = base_url
        self.model = "deepseek-coder:6.7b"
    
    def generate_answer(self, prompt, system_prompt=None):
        """使用本地DeepSeek模型生成回答"""
        payload = {
            "model": self.model,
            "prompt": prompt,
            "system": system_prompt or "你是一个专业的搜索助手。",
            "stream": False,
            "options": {
                "temperature": 0.3,
                "num_predict": 1000
            }
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/api/generate",
                json=payload,
                timeout=60
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"本地模型请求失败: {e}")
            return None
    
    def test_content_visibility(self, query, content_samples):
        """测试内容样本在生成答案中的可见性"""
        results = []
        
        for sample in content_samples:
            # 构建包含内容的提示
            prompt = f"""
            用户问题: {query}
            
            请基于以下参考资料回答问题:
            
            参考资料:
            {sample['content']}
            
            请提供详细的回答，并在适当位置引用信息来源。
            """
            
            answer = self.generate_answer(prompt)
            
            if answer:
                full_response = answer.get('response', '')
                
                # 检查品牌/URL是否被引用
                brand_mentioned = sample.get('brand', '') in full_response
                url_mentioned = sample.get('url', '') in full_response
                
                results.append({
                    "sample_id": sample.get('id'),
                    "query": query,
                    "brand_mentioned": brand_mentioned,
                    "url_mentioned": url_mentioned,
                    "response_preview": full_response[:300]
                })
        
        return results

class ABTestFramework:
    def __init__(self):
        self.tester = LocalDeepSeekTester()
    
    def run_ab_test(self, query, version_a, version_b, iterations=5):
        """运行A/B测试，比较两个内容版本"""
        results_a = []
        results_b = []
        
        for i in range(iterations):
            # 测试版本A
            result_a = self.tester.test_content_visibility(
                query, 
                [{"id": "A", "content": version_a, "brand": "BrandA"}]
            )
            results_a.append(result_a[0])
            
            # 测试版本B
            result_b = self.tester.test_content_visibility(
                query, 
                [{"id": "B", "content": version_b, "brand": "BrandB"}]
            )
            results_b.append(result_b[0])
        
        # 统计分析
        a_reference_rate = sum(1 for r in results_a if r['brand_mentioned']) / iterations
        b_reference_rate = sum(1 for r in results_b if r['brand_mentioned']) / iterations
        
        return {
            "version_a": {
                "reference_rate": a_reference_rate,
                "results": results_a
            },
            "version_b": {
                "reference_rate": b_reference_rate,
                "results": results_b
            },
            "winner": "A" if a_reference_rate > b_reference_rate else "B"
        }

import streamlit as st
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta

class DeepSeekDashboard:
    def __init__(self, data_file):
        self.data_file = data_file
        self.data = self.load_data()
    
    def load_data(self):
        """加载监控数据"""
        # 从文件加载数据
        pass
    
    def render_dashboard(self):
        st.title("DeepSeek GEO监控仪表盘")
        
        # 时间范围选择
        days = st.sidebar.selectbox("时间范围", [7, 14, 30, 90], index=0)
        
        # 总体概览
        st.header("总体概览")
        col1, col2, col3 = st.columns(3)
        
        overall_rate = self.calculate_overall_rate(days)
        col1.metric("总体引用率", f"{overall_rate:.1f}%")
        
        top_keywords = self.get_top_keywords(days)
        col2.metric("表现最佳关键词", top_keywords[0] if top_keywords else "N/A")
        
        trend = self.get_trend(days)
        col3.metric("趋势变化", f"{trend:+.1f}%")
        
        # 关键词引用率图表
        st.header("关键词引用率趋势")
        trend_data = self.get_trend_data(days)
        fig = px.line(trend_data, x='date', y='reference_rate', color='keyword')
        st.plotly_chart(fig)
        
        # 详细数据表格
        st.header("详细数据")
        detail_df = self.get_detail_table(days)
        st.dataframe(detail_df)

# 告警规则示例
ALERT_RULES = {
    "critical_drop": {
        "condition": lambda rate, prev_rate: rate < prev_rate * 0.5,
        "message": "引用率暴跌超过50%",
        "level": "critical"
    },
    "keyword_loss": {
        "condition": lambda rate: rate < 10,
        "message": "关键词引用率低于10%",
        "level": "warning"
    },
    "consistent_decline": {
        "condition": lambda trend: trend < -0.1,
        "message": "连续3天引用率下降",
        "level": "warning"
    }
}

def check_alerts(current_data, historical_data):
    """检查告警规则"""
    alerts = []
    
    for keyword, current_rate in current_data.items():
        prev_rate = historical_data.get(keyword, {}).get('rate', current_rate)
        trend = current_rate - prev_rate
        
        for rule_name, rule in ALERT_RULES.items():
            if rule['condition'](current_rate, prev_rate):
                alerts.append({
                    "keyword": keyword,
                    "rule": rule_name,
                    "message": f"{keyword}: {rule['message']}",
                    "level": rule['level']
                })
    
    return alerts