从数据采集到可视化：基于大模型的微博舆情分析系统实战

最近折腾完一套挺典型的项目基于Python和千问大模型的微博舆情分析与预测系统。简单说就是从微博抓取指定话题的文本做清洗和情感分析再用可视化图表把舆论走势、情感分布、热词演变展示出来顺带把未来几天的情感趋势预测一下。这套东西在毕业设计里属于非常常见的大数据方向选题但真正要做扎实远比想象中麻烦数据采集怎么做才不会封号海量短文本怎么批量交给大模型判断预测结果怎么解释才不显得拍脑袋每一环都有讲究。这篇文章就把完整思路、核心代码、踩过的坑都摊开讲适合正在做相关课题的同学也适合想入门舆情分析这条技术路线的人参考。1. 项目整体设计与方案选型1.1 这个系统到底要解决什么问题微博舆情分析和普通文本分类不一样它面对的是短文本、口语化表达、网络梗、反讽、缩写混杂在一起的真实语料。人工去看一两条没问题但一天几千条甚至几万条微博光靠人眼已经看不完了更不用说还要归纳出整体情绪是偏正面还是偏负面哪些话题在快速发酵未来两天舆情会不会进一步升温这类结论。这个项目要解决的核心问题可以拆成三块。第一数据获取。需要围绕一个关键词或话题持续抓取微博相关内容和传播数据包括正文、发布时间、点赞数、评论数、转发数、地域来源等。第二情感研判。对每条微博判断情感倾向传统做法有用情感词典的也有用机器学习模型的但放在微博场景下精度都不太够。第三可视化呈现。原始数据和分析结果都不能直接扔给用户看要有一个网页界面用图表把趋势、分布、排名、预测结果展示出来让不懂技术的人也能一眼看懂舆论状态。我从一开始就把项目定位成一个完整的小型系统而不是一个孤立的算法Demo。采集、存储、预处理、分析、预测、展示六个环节全部串起来这样无论用于毕业答辩还是实际项目交付逻辑上都站得住。1.2 技术选型为什么是Python 千问大模型技术选型是这个项目最先要定的方向我选择的是Python作为主语言理由非常直接。Python在数据采集、清洗、分析和可视化这条链路里生态极其完整requests、BeautifulSoup可以做采集pandas做数据处理jieba分词Flask搭后端pyecharts或ECharts画图几乎不用切换语言就能跑通全流程。换成其他技术栈光是数据处理环节就得额外折腾。情感分析这部分我对比过几条路线最终选择接入千问大模型API。传统的SnowNLP情感词典方案最简单安装一个库就能用但微博文本里大量存在无语了绝绝子破防了这类网络用语情感词典根本覆盖不到准确率经常在60%上下打转。用朴素贝叶斯或SVM这类机器学习模型又需要人工标注大量训练数据成本太高。千问这类大模型对上下文语义的理解能力强很多像这也太刑了吧这种文本它能结合语境判断出实际是调侃而不是真正的称赞。当然引入大模型也有代价主要是接口调用延迟和费用所以整体设计里我保留了一个缓存层同一段文本不重复调用。这套混合思路既保证了情感判断的准确率又控制了系统运行成本。方案准确率情况开发成本运行成本适用场景SnowNLP情感词典微博场景60%左右极低无快速出Demo机器学习模型依赖标注数据高低有标注资源的团队千问大模型API85%以上低按调用量计费追求效果的个人项目1.3 系统架构五层结构怎么划分这个项目我采用的是分层架构每一层只负责自己的事层与层之间通过数据接口衔接好处是出了问题能快速定位答辩时讲起来也清晰。数据采集层负责调用微博接口或爬虫逻辑把原始微博数据抓回来。数据存储层用MySQL采集到的原始数据和分析结果都落库。数据预处理层做去重、清洗、分词、时间归一化。分析建模层接千问大模型做情感判断再基于时间序列做舆情趋势预测。可视化展示层用Flask加ECharts把分析结果渲染成网页大屏。各层之间我统一用JSON或标准表结构传递数据没有搞复杂的消息队列。对于毕业设计或中小型舆情监控项目引入Kafka、Spark这类重型组件反而会增加部署复杂度没必要。后面如果你想把系统升级成千万级数据量再把数据层换成Spark或者ClickHouse架构上也不冲突。2. 数据采集与预处理舆情分析的底盘2.1 微博数据采集的三种渠道数据采集是整个系统里最容易被低估的一环。很多人一开始觉得爬虫就是发个请求拿返回结果真正做起来才发现微博的登录校验、接口风控、数据字段缺失每一个问题都能卡住大半天。我梳理了三条可行的采集渠道按推荐程度排序。第一条是官方开放接口。通过正规申请流程获取开发者权限用官方API拉取微博数据。它的好处是数据规范、权限清晰、不容易被封缺点是普通个人申请门槛不低接口频率也有限制。如果项目有足够时间提前准备这条路最稳。第二条是网页端爬虫。通过模拟浏览器请求拿到搜索结果页或用户主页的数据。这种方式灵活能拿到官方API不一定开放的字段比如一些互动数据但要做好Cookie维护和请求频率控制。我在项目中主要用这条渠道做增量采集为了回归技术本身的价值相关代码我会做基础演示生产环境请务必评估合规与平台条款。第三条是第三方数据服务商。有些公司专门提供社交媒体数据采集服务花钱买数据最省事。但市面上服务商水平参差不齐数据字段是否齐全、是否稳定更新都不好说而且价格不便宜。除非自己实在没时间研究采集否则我建议优先考虑前两条。2.2 采集代码实现与反爬注意事项我实际用的是网页端搜索接口核心逻辑其实不复杂就是构造搜索请求解析返回的JSON字段入库保存。下面是一个简化版本的数据采集代码演示基本思路。import requests import time import pymysql def fetch_weibo_data(keyword, page1): # 这里只做技术演示请求前必须确认符合平台规则 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Cookie: 你的登录Cookie } params { q: keyword, page: page, sort: time } resp requests.get( https://weibo.com/ajax/statuses/search, paramsparams, headersheaders, timeout10 ) if resp.status_code ! 200: return [] data resp.json() results [] for item in data.get(data, {}).get(list, []): results.append({ weibo_id: item.get(idstr), content: item.get(text_raw, ), user_name: item.get(user, {}).get(screen_name, ), publish_time: item.get(created_at), like_count: item.get(attitudes_count, 0), comment_count: item.get(comments_count, 0), repost_count: item.get(reposts_count, 0) }) return results def save_to_mysql(rows): conn pymysql.connect( hostlocalhost, userroot, password123456, databaseweibo_opinion, charsetutf8mb4 ) cursor conn.cursor() for row in rows: try: cursor.execute( INSERT INTO weibo_posts (weibo_id, content, user_name, publish_time, like_count, comment_count, repost_count) VALUES (%s, %s, %s, %s, %s, %s, %s), (row[weibo_id], row[content], row[user_name], row[publish_time], row[like_count], row[comment_count], row[repost_count]) ) except pymysql.err.IntegrityError: # 重复ID直接跳过 continue conn.commit() cursor.close() conn.close() # 循环采集多页 for page in range(1, 6): rows fetch_weibo_data(深圳天气, page) if not rows: break save_to_mysql(rows) time.sleep(3 page) # 随机延时降低访问频率这里有几个细节值得留意。第一字段尽量用idstr而不要只用id因为某些场景下id会超出JavaScript安全整数范围。第二插入时要处理重复主键否则第二次采集同一话题时会直接报错。第三请求之间的延时我用的是3加页码数相当于每页间隔4到8秒这个频率比较安全。如果你要采集更多页建议加一个指数退避的逻辑遇到频繁被限制就主动拉长间隔。关于反爬我踩过的坑不少核心原则是慢和稳。Cookie会过期最好准备一个Cookie池定时用脚本刷新。IP请求频率太高会被临时限制有条件就上代理池没条件就老老实实把延时加大。不要同时开几十个线程去并发抓取微博的数据量并不是只有一两天能采完长期稳定采集才是关键。2.3 文本清洗与分词原始微博正文里有很多东西是不能直接拿去分析的HTML转义符、URL链接、用户名、话题标签、表情符号、重复的标点。这些噪声如果不清洗会直接干扰情感判断和词频统计。我自己整理了一套清洗流程按顺序执行。import re import jieba import pandas as pd def clean_text(text): if not isinstance(text, str): return # 去除HTML标签和转义符 text re.sub(r[^], , text) text text.replace(nbsp;, ).replace(lt;, ).replace(gt;, ) # 去除URL text re.sub(rhttps?://\S, , text) text re.sub(rwww\.\S, , text) # 去除用户 text re.sub(r[\w\u4e00-\u9fa5\-], , text) # 去除话题标签外层符号保留关键词 text re.sub(r#([^#])#, r\1, text) # 去除多余空白和重复标点 text re.sub(r\s, , text).strip() text re.sub(r([。!?])\1, r\1, text) return text def tokenize(text): words jieba.lcut(text) stopwords {的, 了, 是, 我, 你, 他, 这, 那, 也, 就, 都, 而, 及, 与, 或} result [] for word in words: word word.strip() if word and word not in stopwords and len(word) 1: result.append(word) return result # 清洗分词 df pd.read_sql(SELECT content FROM weibo_posts LIMIT 5000, conn) df[clean_content] df[content].apply(clean_text) df[words] df[clean_content].apply(tokenize)清洗有一个容易被忽略的问题文本清洗后可能变成空字符串。比如一条微博全是表情和URL清洗完就什么都没有了。这种空文本必须过滤掉否则后面调用大模型接口时白白消耗一次请求。过滤规则建议是清洗后长度小于2的直接丢弃。分词这块jieba默认词库对网络热词的支持不够像破防蚌埠住了尊嘟假嘟会被切成奇怪的碎片。我通常会自定义一个词典文件把这些网络热词整体加入分词效果会好很多。情感分析之前的词云展示和热门话题提取都依赖分词质量这一步值得花时间调。3. 千问大模型情感分析与预测实现3.1 为什么用大模型替代传统情感词典情感分析是这套系统的技术亮点也是最容易说出东西的部分。我刚开始做的时候试过SnowNLP接口很简单几行代码就能算出一条文本的情感分数但效果放在微博语料上确实不够看。举个例子今天这操作真的牛这句话SnowNLP会判定为正面因为它看到牛这个字天然带褒义。但如果这条微博配的是一段翻车视频实际表达的是讽刺SnowNLP就会判错。还有太刑了有判头这类谐音梗传统词典更是完全没有办法。千问大模型能结合整句话的语境去理解知道这些词在具体场景里是讽刺还是夸赞判断的准确率高一大截。当然大模型不是万能的。对于特别长的文本或者夹杂大量图片描述、无意义刷屏内容偶尔也会判偏。所以在设计里我加了一个置信度的概念让模型不仅告诉我情感倾向还告诉我有多大的把握低置信度的结果不会被直接采用。3.2 Prompt设计与批量调用实现调用千问大模型做情感分析最核心的是Prompt设计。Prompt写得好不好直接决定输出格式规不规范、判断准不准。我试过两种写法第一种是自由描述式请判断这条微博是正面还是负面返回结果五花八门有的带解释有的带感叹号解析起来很痛苦。第二种是强制JSON输出式效果稳定得多。import requests import json import time PROMPT_TEMPLATE 你是一名社交媒体舆情分析专家。 请对下面这条微博文本进行情感判断只输出JSON不要输出任何其他内容。 JSON格式要求如下 {sentiment: positive, confidence: 0.95, reason: 简要原因} 其中sentiment只能取 positive 或 neutral 或 negative。 微博文本{text} def analyze_with_qwen(text, api_key): api_url https://api.qwen.example.com/v1/chat/completions # 上面的URL仅作格式演示实际以你的模型平台公布地址为准 headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { model: qwen-turbo, messages: [ {role: system, content: 你只输出JSON格式结果。}, {role: user, content: PROMPT_TEMPLATE.format(texttext)} ], temperature: 0.1, max_tokens: 200 } resp requests.post(api_url, jsonpayload, headersheaders, timeout15) resp.raise_for_status() content resp.json()[choices][0][message][content] # 容错清理可能出现的多余字符 content content.strip().strip(json).strip().strip() result json.loads(content) return result # 批量处理示例 df pd.read_sql(SELECT id, clean_content FROM weibo_posts WHERE sentiment_label IS NULL LIMIT 100, conn) api_key 你的APIKey for idx, row in df.iterrows(): try: result analyze_with_qwen(row[clean_content], api_key) cursor.execute( UPDATE weibo_posts SET sentiment_label%s, sentiment_score%s WHERE id%s, (result[sentiment], result[confidence], row[id]) ) except Exception as e: print(fID {row[id]} 分析失败: {e}) continue # 控制请求频率避免触发接口限流 time.sleep(0.5) conn.commit()这里有几个经验值得重点说明。温度参数我设置的是0.1温度越低输出越稳定不会出现同一句话两次判断结果不一样的情况。max_tokens设置成200就够用了因为要求只输出JSON结构不需要模型长篇大论。最关键的是在Prompt里写死了JSON字段名和取值的枚举范围这样解析结果就不会出错。关于批量调用我强烈建议在本地加一层缓存。可以建一张分析结果表以文本的MD5值为唯一键每次调用前先查缓存命中就直接用不重复调接口。我实际测试过真实舆情数据里重复或高度相似的文本占比不低缓存能省下20%到30%的调用量长期跑下来费用差距很明显。3.3 情感量化与舆情指数计算模型输出的是positive、neutral、negative三个类别和对应的置信度但画折线图需要的是一个连续值不能直接把字符串塞给ECharts。我定义了一个叫情感指数的指标用来描述某一天的舆论整体偏向。计算方式并不复杂。正面记1分中性记0分负面记-1分然后用模型的置信度作为权重再做平均。label_map {positive: 1, neutral: 0, negative: -1} def cal_sentiment_score(df): # df 包含 sentiment_label 和 sentiment_confidence 两列 total_weight 0.0 total_score 0.0 for _, row in df.iterrows(): base label_map.get(row[sentiment_label], 0) weight row.get(sentiment_confidence, 0.5) total_score base * weight total_weight weight if total_weight 0: return 0 return round(total_score / total_weight, 4)按天分组计算就能得到一条情感指数的时间序列。情感指数取值范围是-1到1大于0说明当天舆论偏向正面小于0说明偏向负面。同时我还会统计每天的微博总量、正面比例、负面比例这些字段在可视化大屏里会用到。这里有一个容易踩坑的点不要直接用0.5作为中性类别的分母权重。中性文本其实传递的舆论信号很弱如果给它过高的权重会把整体情感指数往0方向拉导致趋势看起来平平无奇。我的做法是将中性文本的权重乘以0.3降低它对整体指数的贡献这样曲线对真实舆论变化的反应更灵敏。3.4 舆情趋势预测的两种做法预测模块是整个系统里比较有技术含量的一部分也是答辩时经常被深挖的地方。我的项目目标不是预测具体的某条微博会不会火而是预测未来3天的情感指数和讨论热度走势。这属于典型的时间序列预测问题。我先用了经典的时间序列模型ARIMA理由是可解释性强、代码轻量、十几行就能跑起来。ARIMA需要调三个参数p、d、q我一般先做差分让序列平稳再根据自相关图和偏自相关图来粗选参数。下面的代码展示了一个基础流程。from statsmodels.tsa.arima.model import ARIMA import pandas as pd # daily_df 包含 date 和 sentiment_score 两列按天汇总 series daily_df.set_index(date)[sentiment_score].astype(float) # 一阶差分后序列更平稳 model ARIMA(series, order(2, 1, 2)) model_fit model.fit() # 预测未来3天 forecast model_fit.forecast(steps3) print(forecast)如果数据量足够也可以上LSTM做多步预测。LSTM在毕业设计里更唬人但实际效果对数据量要求很高几十天的数据根本训练不出什么。我在项目里的策略是预测部分用ARIMA作为主方案同时做一个简单的线性回归基准模型两者结果做对比展示。这样答辩时既能讲经典模型又能讲深度学习方向的尝试显得思考更有层次。做预测一定要有边界意识。微博舆情受突发事件影响极大今天的一条热搜明天可能就反转任何模型都不可能精准预测这类突变。我在系统里把预测结果用虚线展示与历史实线区分开界面上也会标注预测结果仅作趋势参考。这样既专业也避免用户把预测当成确定性结论。4. 可视化与系统集成4.1 大屏页面应该包含哪些图表可视化是舆情分析系统的门面用户第一眼看到的就是大屏效果。如果图表设计得乱模型再准也白搭。我根据舆情分析的业务场景规划了五个核心图表模块。顶部是总览指标卡显示监控关键词、累计微博数、情感指数、正面比例、负面比例。左侧是情感趋势折线图展示最近30天情感指数变化并把未来3天预测值用虚线接上。中间区域放词云展示当前周期内的高频关键词词语越大说明热度越高这也是最能直观展示舆情焦点的模块。右侧放热门话题Top10柱状图我通过短语提取和计数统计得到当前讨论度最高的话题短语。底部放一条时间热度分布图展示一天24小时内的微博发布规律。这些图表全部用ECharts渲染。ECharts配置项很灵活我前后端只用JSON传数据后端pandas算好数据后直接转成前端可用的格式前端通过Ajax拉取渲染成图表。4.2 Flask接口设计与数据库表结构系统后端我选了Flask因为它轻量路由简单适合把数据库查询结果直接转成JSON接口。数据库表结构方面原始微博数据表weibo_posts是核心另外还有一张敏感词表、一张系统配置表用来存监控关键词和采集状态。下面是最核心的建表语句字段设计我讲一下理由。CREATE TABLE weibo_posts ( id INT AUTO_INCREMENT PRIMARY KEY, weibo_id VARCHAR(64) UNIQUE, content TEXT, clean_content TEXT, user_name VARCHAR(64), publish_time DATETIME, like_count INT DEFAULT 0, comment_count INT DEFAULT 0, repost_count INT DEFAULT 0, location VARCHAR(64), sentiment_label VARCHAR(16), sentiment_confidence FLOAT, sentiment_score FLOAT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_publish_time (publish_time), INDEX idx_sentiment_label (sentiment_label) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;weibo_id字段加唯一约束是为了防止爬虫重复插入数据。clean_content单独存一列避免每次做情感分析都要重新清洗。sentiment_label和sentiment_confidence刚开始可能为空等大模型分析完再回填这样采集和分析流程可以解耦先跑采集任务再慢慢批量分析。索引方面publish_time和sentiment_label是查询高频字段必须加索引否则数据量上来之后接口会明显变慢。Flask接口设计的套路很简单就是查数据库组装JSON返回给前端。from flask import Flask, jsonify import pymysql import pandas as pd app Flask(__name__) def query_df(sql): conn pymysql.connect( hostlocalhost, userroot, password123456, databaseweibo_opinion, charsetutf8mb4 ) df pd.read_sql(sql, conn) conn.close() return df app.route(/api/sentiment_trend) def sentiment_trend(): df query_df( SELECT DATE(publish_time) AS d, AVG(sentiment_score) AS score FROM weibo_posts WHERE sentiment_score IS NOT NULL GROUP BY DATE(publish_time) ORDER BY d ) return jsonify({ dates: df[d].astype(str).tolist(), scores: df[score].round(4).tolist() }) app.route(/api/keywords_cloud) def keywords_cloud(): # 实际实现需先从库里取文本做分词统计 words query_df(SELECT clean_content FROM weibo_posts LIMIT 5000) counter {} for content in words[clean_content].dropna(): for w in tokenize(content): counter[w] counter.get(w, 0) 1 top_words sorted(counter.items(), keylambda x: x[1], reverseTrue)[:50] return jsonify({items: [{name: w, value: c} for w, c in top_words]}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)接口返回的时间字段我会提前转成字符串避免pandas的Timestamp类型无法被JSON序列化这个问题很多人第一次都会遇到。4.3 定时增量采集与系统自动化舆情监控系统不能只跑一次就完事需要能够每天自动增量采集数据。我用了APScheduler库做定时任务配置了两种任务。第一种是采集任务每天固定时间点运行比如每天早上八点和晚上八点各抓取一次。第二种是情感分析任务采集完成后自动对未分析的文本调用大模型接口。通过设置了一个状态字段来区分哪些文本已经分析过、哪些还没有避免重复计算。from apscheduler.schedulers.blocking import BlockingScheduler def collect_job(): print(开始增量采集) rows fetch_weibo_data(监控关键词, page1) save_to_mysql(rows) def analyze_job(): print(开始情感分析回填) # 从库里取未分析的文本调用千问大模型更新情感字段 pass scheduler BlockingScheduler(timezoneAsia/Shanghai) scheduler.add_job(collect_job, cron, hour8) scheduler.add_job(collect_job, cron, hour20) scheduler.add_job(analyze_job, cron, hour9) scheduler.add_job(analyze_job, cron, hour21) scheduler.start()定时任务部署的时候有一个坑跑在开发环境里没问题但换到服务器上如果进程挂了没人知道。建议搭配日志文件把每次任务的执行结果追加写入日志出问题能快速排查。另外如果你用的是单核的小服务器定时任务和大模型批量分析不能同时跑内存和CPU都会吃紧最好错开执行时间。5. 常见问题与避坑清单5.1 高频问题速查表这套系统从开发到测试跑通前后踩了不少坑。我把高频问题整理成一张速查表方便你遇到问题时直接对照排查。问题现象可能原因解决办法微博采集几次后被限制Cookie失效或请求频率过高准备Cookie池增大延时降低并发插入数据库总是报主键冲突weibo_id重复使用INSERT IGNORE或捕获IntegrityError大模型接口返回超时网络波动或单次请求token太长加timeout重试缩短文本长度分批调用情感分析结果全是neutralPrompt没有约束分类倾向调整Prompt明确要求结合语境判断降低temperature到0.1JSON解析大模型返回失败返回内容带有markdown格式清理json标记后再json.loadsECharts图表中文乱码后端返回编码不是UTF-8设置response头为application/json; charsetutf-8ARIMA预测结果是一条直线原始序列方差过大或参数不合适对序列做差分尝试不同p、q参数词云里全是无意义的高频词停用词表覆盖不够扩充停用词加入网络泛化词如哈哈哈、谢谢Flask接口访问慢全表扫描没有走索引给查询字段加索引限制返回条数5.2 踩坑实录与排查思路这里分享两个我觉得最有代表性的排查过程。第一个是情感分析结果标题显示乱码。排查下来发现Flask返回的默认Content-Type没有指定charset导致浏览器用默认编码解析中文JSON出现乱码。解决办法是在Flask的Response对象上显式设置content_type为application/json; charsetutf-8或者在jsonify之后统一添加响应头。这个问题不是特别难但很影响演示效果最早做出来大屏一片乱码看着就慌。第二个是批量调用大模型时被限流。我最早用多线程并发调用100条数据开10个线程看上去效率很高结果跑了几十条之后接口开始报429错误。大模型开放接口普遍有针对账号的QPS限制盲目增加并发只会更快触发限制。后来改成单线程逐条调用每条之间间隔0.5秒反而稳定跑完了几千条数据的分析。我还想特别提醒一点预测模块的结果一定要在界面上解释清楚。很多用户看到预测曲线会误以为系统能精准预测舆情走向实际上模型只能基于历史规律做趋势外推突发事件一来预测立即失效。我把预测结果统一标成虚线并在页面底部加了一段说明文字告诉用户预测结果受外界因素影响较大仅作参考。这样既保持系统的严谨性也避免后续交付过程中出现不必要的争议。另外一个容易被忽视的坑是数据量增长后的查询性能。我只采集了不到一周的数据时接口响应挺快没什么感觉。但跑到一个月后weibo_posts表里十几万条数据不带索引的情感趋势查询开始变慢前端图表要等好几秒才能出来。后来加了publish_time和sentiment_label的联合索引查询速度明显改善。如果你的数据规模再上一个量级聚合统计可以考虑用缓存表每天定时把统计结果算好存起来前端直接读结果就行。最后分享一个我后来才体会到的经验。做这种舆情分析项目技术本身是一方面更重要的是对数据的理解。刚开始我只关注情感分析准确率觉得模型结果准就够了后来发现如果采集到的数据本身偏向性很强或者某个时间段微博量特别稀疏再好的算法也救不回来。做数据采集时就应该多花点时间确认数据覆盖是否均匀、字段是否完整这才是整个系统的地基。这套项目做完之后我对数据质量决定模型上限这句话的理解深了不少也希望这篇文章能帮你绕开我走过的弯路。