【营销】竞品分析智能体：爬取网页，生成对比表格与策略

1. 需求分析与技术架构

1.1 核心需求

竞品分析智能体需要具备以下核心能力：

• 网页爬取：自动爬取指定竞品网站的相关信息
• 数据提取：从爬取的网页中提取关键产品信息和特性
• 数据分析：分析和对比不同竞品的优缺点
• 表格生成：生成结构化的竞品对比表格
• 策略建议：基于分析结果提供营销策略建议
• 可视化输出：展示关键对比数据的图表
• 报告生成：生成完整的竞品分析报告

1.2 技术架构

我们将采用以下技术栈构建竞品分析智能体：

┌─────────────────────┐
│     用户界面层      │
│  Gradio Web界面     │
└──────────┬──────────┘
           │
┌──────────▼──────────┐
│     智能体核心层     │
│  LangChain + LLM    │
└──────────┬──────────┘
           │
┌──────────▼──────────┐
│     工具与数据层     │
│ 1. 网页爬取工具      │
│ 2. 数据提取工具      │
│ 3. 数据分析工具      │
│ 4. 报告生成工具      │
└─────────────────────┘

2. 环境搭建与依赖配置

首先，我们需要创建项目并安装必要的依赖：

# 创建项目目录
mkdir competitor-analysis-agent
cd competitor-analysis-agent

# 初始化Python环境
python -m venv venv
venv\Scripts\activate  # Windows
# 或 source venv/bin/activate  # macOS/Linux

# 安装依赖
pip install langchain langchain-openai gradio requests beautifulsoup4 pandas numpy matplotlib seaborn

3. 核心功能实现

3.1 配置文件管理

创建 config.py 文件管理配置信息：

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

# API密钥配置
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# 系统配置
SYSTEM_PROMPT = """你是一个专业的市场分析师，负责进行竞品分析并提供营销策略建议。

你的职责包括：
1. 分析爬取的竞品信息，提取关键数据点
2. 对比不同竞品的优缺点和特性
3. 识别市场趋势和机会
4. 生成结构化的竞品对比表格
5. 基于分析结果提供具体的营销策略建议
6. 生成完整、专业的竞品分析报告

请保持客观、专业的分析态度，确保分析结果有数据支持，建议具有可操作性。"""

# 爬取配置
CRAWLER_CONFIG = {
    "headers": {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    },
    "timeout": 30,
    "max_retries": 3
}

# 竞品分析维度配置
ANALYSIS_DIMENSIONS = {
    "产品特性": ["核心功能", "技术参数", "用户体验", "设计风格"],
    "价格策略": ["基础定价", "套餐方案", "折扣策略", "价值主张"],
    "市场定位": ["目标用户", "市场份额", "品牌形象", "差异化优势"],
    "营销渠道": ["线上渠道", "线下渠道", "社交媒体", "内容营销"],
    "客户服务": ["支持方式", "响应时间", "服务质量", "用户评价"]
}

3.2 网页爬取工具

创建 web_crawler.py 文件实现网页爬取：

# web_crawler.py
import requests
from bs4 import BeautifulSoup
import time
from config import CRAWLER_CONFIG

class WebCrawler:
    """网页爬取类"""
    
    def __init__(self):
        self.headers = CRAWLER_CONFIG["headers"]
        self.timeout = CRAWLER_CONFIG["timeout"]
        self.max_retries = CRAWLER_CONFIG["max_retries"]
    
    def crawl_page(self, url):
        """爬取单个网页"""
        for retry in range(self.max_retries):
            try:
                response = requests.get(url, headers=self.headers, timeout=self.timeout)
                if response.status_code == 200:
                    return response.text
                else:
                    print(f"爬取失败，状态码：{response.status_code}")
            except Exception as e:
                print(f"爬取出错：{str(e)}")
            
            # 重试间隔
            time.sleep(2)
        
        return None
    
    def extract_content(self, html, selectors=None):
        """从HTML中提取内容"""
        soup = BeautifulSoup(html, "html.parser")
        
        if selectors:
            # 使用指定的选择器提取内容
            content = {}
            for key, selector in selectors.items():
                elements = soup.select(selector)
                if elements:
                    content[key] = [element.get_text(strip=True) for element in elements]
                else:
                    content[key] = []
            return content
        else:
            # 提取所有文本内容
            return soup.get_text(strip=True)
    
    def crawl_multiple_pages(self, urls):
        """爬取多个网页"""
        results = {}
        
        for url in urls:
            print(f"正在爬取：{url}")
            html = self.crawl_page(url)
            if html:
                results[url] = html
            # 避免过快请求
            time.sleep(1)
        
        return results
    
    def crawl_product_info(self, product_urls):
        """爬取产品信息"""
        product_info = {}
        
        for product_name, url in product_urls.items():
            print(f"正在爬取产品：{product_name}")
            html = self.crawl_page(url)
            if html:
                # 提取基本产品信息
                soup = BeautifulSoup(html, "html.parser")
                
                # 尝试提取标题
                title = soup.find("title").get_text(strip=True) if soup.find("title") else ""
                
                # 尝试提取元描述
                meta_desc = ""
                meta_desc_tag = soup.find("meta", attrs={"name": "description"})
                if meta_desc_tag:
                    meta_desc = meta_desc_tag.get("content", "")
                
                # 尝试提取正文内容
                body_content = ""
                body = soup.find("body")
                if body:
                    # 移除脚本和样式
                    for script in body(["script", "style"]):
                        script.decompose()
                    body_content = body.get_text(separator="\n", strip=True)
                
                product_info[product_name] = {
                    "url": url,
                    "title": title,
                    "meta_description": meta_desc,
                    "content": body_content
                }
            
            # 避免过快请求
            time.sleep(1)
        
        return product_info

3.3 数据分析工具

创建 data_analyzer.py 文件实现数据分析：

# data_analyzer.py
from langchain_openai import ChatOpenAI
from config import OPENAI_API_KEY, ANALYSIS_DIMENSIONS
import pandas as pd

class DataAnalyzer:
    """数据分析类"""
    
    def __init__(self):
        self.llm = ChatOpenAI(
            api_key=OPENAI_API_KEY,
            model="gpt-3.5-turbo",
            temperature=0.3
        )
    
    def analyze_product(self, product_name, product_data):
        """分析单个产品"""
        prompt = f"""
请分析以下产品信息，提取关键特性和优势：

产品名称：{product_name}
产品URL：{product_data['url']}
产品标题：{product_data['title']}
元描述：{product_data['meta_description']}
产品内容：{product_data['content'][:2000]}...

请按照以下维度分析：
{ANALYSIS_DIMENSIONS}

每个维度请提供具体的分析结果，确保分析基于提供的产品信息。
        """
        
        response = self.llm.invoke(prompt)
        return response.content
    
    def compare_products(self, product_analyses):
        """对比多个产品"""
        analyses_text = "\n\n".join([f"产品：{name}\n分析：{analysis}" for name, analysis in product_analyses.items()])
        
        prompt = f"""
请对比以下产品分析结果，生成详细的竞品对比：

{analyses_text}

对比要求：
1. 按照{ANALYSIS_DIMENSIONS.keys()}维度进行对比
2. 分析每个产品的优缺点
3. 识别市场空白和机会
4. 生成结构化的对比结果

请确保对比分析客观、全面，基于提供的产品信息。
        """
        
        response = self.llm.invoke(prompt)
        return response.content
    
    def generate_comparison_table(self, comparison_result):
        """生成对比表格"""
        prompt = f"""
请根据以下竞品对比结果，生成结构化的对比表格：

{comparison_result}

表格要求：
1. 使用Markdown格式
2. 包含产品名称和各个分析维度
3. 每个维度下包含具体的对比内容
4. 表格清晰易读

请直接生成表格，不要包含其他无关内容。
        """
        
        response = self.llm.invoke(prompt)
        return response.content
    
    def generate_strategy(self, comparison_result, company_name):
        """生成营销策略建议"""
        prompt = f"""
基于以下竞品对比结果，为{company_name}提供详细的营销策略建议：

{comparison_result}

建议要求：
1. 基于竞品分析结果
2. 识别差异化机会
3. 提供具体的营销策略
4. 包括产品定位、价格策略、营销渠道等方面
5. 建议具有可操作性

请确保建议基于数据支持，切实可行。
        """
        
        response = self.llm.invoke(prompt)
        return response.content
    
    def generate_report(self, product_analyses, comparison_result, strategy, company_name):
        """生成完整的竞品分析报告"""
        report = f"# 竞品分析报告\n\n"
        report += f"## 报告概览\n"
        report += f"- 分析时间：2024-01-01\n"
        report += f"- 分析公司：{company_name}\n"
        report += f"- 分析竞品：{', '.join(product_analyses.keys())}\n\n"
        
        report += "## 竞品分析\n"
        for product_name, analysis in product_analyses.items():
            report += f"### {product_name}\n"
            report += f"{analysis}\n\n"
        
        report += "## 竞品对比\n"
        report += f"{comparison_result}\n\n"
        
        report += "## 营销策略建议\n"
        report += f"{strategy}\n\n"
        
        report += "## 结论\n"
        report += "基于以上分析，我们建议...\n"
        
        return report
    
    def create_comparison_dataframe(self, comparison_table):
        """将对比表格转换为DataFrame"""
        # 这里需要根据实际生成的表格格式进行解析
        # 简化实现，实际应用中可能需要更复杂的解析逻辑
        import re
        
        # 提取表格数据
        table_rows = re.findall(r'\|(.*?)\|', comparison_table)
        if not table_rows:
            return None
        
        # 解析表头
        header = [cell.strip() for cell in table_rows[0].split('|') if cell.strip()]
        
        # 解析数据行
        data = []
        for row in table_rows[2:]:  # 跳过表头和分隔线
            cells = [cell.strip() for cell in row.split('|') if cell.strip()]
            if cells:
                data.append(cells)
        
        # 创建DataFrame
        df = pd.DataFrame(data, columns=header)
        return df

3.4 智能体构建

创建 agent.py 文件构建智能体：

# agent.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType, initialize_agent
from langchain.memory import ConversationBufferMemory
from web_crawler import WebCrawler
from data_analyzer import DataAnalyzer
from config import OPENAI_API_KEY, SYSTEM_PROMPT

# 初始化LLM
llm = ChatOpenAI(
    api_key=OPENAI_API_KEY,
    model="gpt-3.5-turbo",
    temperature=0.7
)

# 初始化记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 初始化工具类
web_crawler = WebCrawler()
data_analyzer = DataAnalyzer()

# 定义智能体工具
from langchain.tools import tool

@tool

def crawl_competitor_info(product_urls: str) -> str:
    """爬取竞品信息"""
    try:
        import json
        urls = json.loads(product_urls)
        product_info = web_crawler.crawl_product_info(urls)
        return json.dumps(product_info, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"爬取竞品信息时出错：{str(e)}"

@tool

def analyze_products(product_info: str) -> str:
    """分析产品信息"""
    try:
        import json
        info = json.loads(product_info)
        
        # 分析每个产品
        product_analyses = {}
        for product_name, data in info.items():
            analysis = data_analyzer.analyze_product(product_name, data)
            product_analyses[product_name] = analysis
        
        # 对比产品
        comparison = data_analyzer.compare_products(product_analyses)
        
        # 生成对比表格
        comparison_table = data_analyzer.generate_comparison_table(comparison)
        
        return json.dumps({
            "product_analyses": product_analyses,
            "comparison": comparison,
            "comparison_table": comparison_table
        }, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"分析产品时出错：{str(e)}"

@tool

def generate_strategy(comparison_result: str, company_name: str) -> str:
    """生成营销策略"""
    try:
        import json
        result = json.loads(comparison_result)
        comparison = result.get("comparison", "")
        
        # 生成营销策略
        strategy = data_analyzer.generate_strategy(comparison, company_name)
        
        # 生成完整报告
        report = data_analyzer.generate_report(
            result.get("product_analyses", {}),
            comparison,
            strategy,
            company_name
        )
        
        return json.dumps({
            "strategy": strategy,
            "report": report
        }, ensure_ascii=False, indent=2)
    except Exception as e:
        return f"生成策略时出错：{str(e)}"

# 定义工具列表
tools = [
    crawl_competitor_info,
    analyze_products,
    generate_strategy
]

# 初始化智能体
competitor_agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    system_message=SYSTEM_PROMPT,
    verbose=True
)

3.5 Web界面构建

创建 app.py 文件构建Gradio Web界面：

# app.py
import gradio as gr
import json
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
from agent import competitor_agent
from data_analyzer import DataAnalyzer

# 设置中文字体
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

# 处理竞品爬取
def handle_crawl_competitors(product_urls_text):
    # 解析输入的URLs
    try:
        urls = {}
        for line in product_urls_text.strip().split('\n'):
            if '|' in line:
                name, url = line.split('|', 1)
                urls[name.strip()] = url.strip()
        
        import json
        urls_json = json.dumps(urls, ensure_ascii=False)
        
        # 调用智能体爬取竞品信息
        result = competitor_agent.invoke(f"爬取以下竞品信息：{urls_json}")
        return result['output']
    except Exception as e:
        return f"爬取竞品信息时出错：{str(e)}"

# 处理产品分析
def handle_analyze_products(product_info):
    # 调用智能体分析产品
    result = competitor_agent.invoke(f"分析以下产品信息：{product_info}")
    return result['output']

# 处理策略生成
def handle_generate_strategy(comparison_result, company_name):
    # 调用智能体生成策略
    result = competitor_agent.invoke(f"基于以下对比结果为{company_name}生成营销策略：{comparison_result}")
    return result['output']

# 生成对比图表
def generate_comparison_chart(comparison_result):
    try:
        import json
        result = json.loads(comparison_result)
        comparison_table = result.get("comparison_table", "")
        
        # 使用DataAnalyzer解析表格
        analyzer = DataAnalyzer()
        df = analyzer.create_comparison_dataframe(comparison_table)
        
        if df is None:
            return "无法生成图表：未找到有效的对比表格"
        
        # 生成简单的对比图表
        plt.figure(figsize=(12, 8))
        
        # 假设我们有一个评分列
        if '评分' in df.columns:
            # 转换评分为数字
            df['评分'] = pd.to_numeric(df['评分'], errors='coerce')
            sns.barplot(x='产品', y='评分', data=df)
            plt.title('竞品评分对比')
        else:
            # 生成产品特性对比
            # 这里需要根据实际表格内容进行调整
            sns.countplot(x='产品', data=df)
            plt.title('竞品特性对比')
        
        plt.xticks(rotation=45, ha='right')
        plt.tight_layout()
        
        # 保存图表
        chart_path = "competitor_comparison.png"
        plt.savefig(chart_path)
        plt.close()
        
        return chart_path
    except Exception as e:
        return f"生成图表时出错：{str(e)}"

# 创建Gradio界面
with gr.Blocks(title="竞品分析智能助手") as demo:
    gr.Markdown("""
    # 竞品分析智能助手
    输入竞品URL，自动爬取信息并生成分析报告！
    """)
    
    # 标签页
    with gr.Tabs():
        # 竞品爬取标签页
        with gr.TabItem("竞品爬取"):
            urls_input = gr.Textbox(label="输入竞品信息（格式：产品名称 | URL，每行一个）", lines=10, placeholder="例如：\n产品A | https://example.com/product-a\n产品B | https://example.com/product-b")
            crawl_btn = gr.Button("开始爬取")
            crawl_output = gr.Textbox(label="爬取结果", lines=10)
            
            crawl_btn.click(
                fn=handle_crawl_competitors,
                inputs=[urls_input],
                outputs=[crawl_output]
            )
        
        # 产品分析标签页
        with gr.TabItem("产品分析"):
            product_info_input = gr.Textbox(label="产品信息", lines=10, placeholder="请粘贴爬取结果...")
            analyze_btn = gr.Button("分析产品")
            analyze_output = gr.Textbox(label="分析结果", lines=15)
            chart_output = gr.Image(label="对比图表")
            
            def analyze_and_generate_chart(product_info):
                analysis_result = handle_analyze_products(product_info)
                chart = generate_comparison_chart(analysis_result)
                return analysis_result, chart
            
            analyze_btn.click(
                fn=analyze_and_generate_chart,
                inputs=[product_info_input],
                outputs=[analyze_output, chart_output]
            )
        
        # 策略生成标签页
        with gr.TabItem("策略生成"):
            comparison_input = gr.Textbox(label="对比结果", lines=10, placeholder="请粘贴分析结果...")
            company_input = gr.Textbox(label="公司名称", placeholder="请输入你的公司名称...")
            strategy_btn = gr.Button("生成策略")
            strategy_output = gr.Textbox(label="策略建议", lines=20)
            
            strategy_btn.click(
                fn=handle_generate_strategy,
                inputs=[comparison_input, company_input],
                outputs=[strategy_output]
            )

# 启动应用
if __name__ == "__main__":
    demo.launch(share=True)

4. 部署与使用

4.1 本地部署

# 运行应用
python app.py

4.2 Docker容器化

创建 Dockerfile 文件：

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 安装中文字体
RUN apt-get update && apt-get install -y fonts-wqy-zenhei

COPY . .

EXPOSE 7860

CMD ["python", "app.py"]

创建 requirements.txt 文件：

langchain
langchain-openai
gradio
requests
beautifulsoup4
pandas
numpy
matplotlib
seaborn
python-dotenv

4.3 使用流程

1. 竞品输入：在"竞品爬取"标签页输入竞品名称和URL
2. 开始爬取：点击"开始爬取"按钮，智能体将爬取竞品信息
3. 分析产品：在"产品分析"标签页粘贴爬取结果，点击"分析产品"按钮
4. 查看图表：查看自动生成的竞品对比图表
5. 生成策略：在"策略生成"标签页粘贴分析结果，输入公司名称，点击"生成策略"按钮
6. 查看报告：查看智能体生成的营销策略建议和完整分析报告

5. 功能测试与优化

5.1 测试场景

1. 简单竞品分析：

   • 输入2-3个竞品的URL
   • 验证智能体是否能正确爬取信息并生成分析
   • 检查对比表格和策略建议的质量

2. 复杂竞品分析：

   • 输入5个以上竞品的URL
   • 验证智能体是否能处理大量数据并生成全面分析
   • 检查分析的深度和洞察力

3. 不同行业测试：

   • 测试不同行业（科技、零售、服务等）的竞品分析
   • 验证智能体对不同行业特性的理解
   • 检查行业特定的分析维度是否被覆盖

4. 爬取能力测试：

   • 测试爬取不同类型网站（静态、动态）的能力
   • 验证智能体对反爬机制的应对能力
   • 检查数据提取的准确性

5.2 优化建议

1. 爬取能力优化：

   • 增加对动态网站的爬取能力（如使用Selenium）
   • 实现更智能的反爬机制（如随机User-Agent、IP代理等）
   • 优化数据提取逻辑，提高信息提取的准确性

2. 分析能力增强：

   • 增加更多分析维度，如技术架构、团队背景等
   • 实现情感分析，分析用户对竞品的评价情感
   • 增加时间序列分析，跟踪竞品的发展变化

3. 报告质量提升：

   • 增加更多可视化图表类型（雷达图、热力图等）
   • 实现报告的自动排版和美化
   • 支持报告导出为PDF、Word等格式

4. 用户体验优化：

   • 增加爬取进度显示
   • 实现分析过程的实时反馈
   • 优化界面设计，提高操作流畅性

5. 功能扩展：

   • 增加竞品监测功能，定期自动爬取和分析
   • 实现竞品价格监控和预警
   • 集成市场趋势分析，结合行业数据

6. 总结与展望

6.1 项目总结

本实战案例成功构建了一个竞品分析智能体，具备以下核心功能：

• ✅ 网页自动爬取与信息提取
• ✅ 多维度产品分析
• ✅ 结构化竞品对比表格生成
• ✅ 数据驱动的营销策略建议
• ✅ 可视化对比图表
• ✅ 完整的竞品分析报告生成
• ✅ Web界面交互
• ✅ Docker容器化部署

6.2 未来展望

1. 智能化升级：

   • 引入机器学习模型，提高数据提取和分析的准确性
   • 使用自然语言处理技术，增强对非结构化数据的理解
   • 开发预测模型，预测竞品的未来发展趋势

2. 功能扩展：

   • 支持社交媒体数据爬取和分析
   • 集成搜索引擎数据，分析竞品的搜索表现
   • 实现用户评论分析，了解用户对竞品的真实评价

3. 平台集成：

   • 与营销自动化平台集成
   • 与CRM系统集成，提供客户视角的竞品分析
   • 与项目管理工具集成，支持团队协作

4. 技术创新：

   • 探索使用本地大模型进行分析，保护数据隐私
   • 实现分布式爬取，提高大规模竞品分析的效率
   • 开发智能代理网络，应对复杂的反爬机制

通过本项目的实践，我们不仅掌握了AI智能体在市场营销领域的应用方法，也了解了从网页爬取到策略生成的完整技术流程。随着技术的不断进步，竞品分析智能体将在市场营销领域发挥越来越重要的作用，为企业提供更全面、更深入的市场洞察，帮助企业制定更有效的竞争策略。