登陆网站密码不保存怎么做,wordpress-move,吉林响应式网站建设,企业营销型网站建设公司多智能体协作的自动化视频舆情分析报告生成器
1. 项目的意义与价值
从“非结构化视频”中挖掘“结构化洞察”的通用挑战
在当今的数字生态中#xff0c;视频已成为信息传播、知识分享和消费者意见表达的核心媒介。从企业内部的会议录屏、技术培训#xff0c;到外部的市场宣传…多智能体协作的自动化视频舆情分析报告生成器
1. 项目的意义与价值
从“非结构化视频”中挖掘“结构化洞察”的通用挑战
在当今的数字生态中视频已成为信息传播、知识分享和消费者意见表达的核心媒介。从企业内部的会议录屏、技术培训到外部的市场宣传、用户评测海量视频内容正以前所未有的速度被创造出来。然而对于绝大多数组织而言这个庞大的视频库仍然是“暗数据”——蕴含着无尽的价值却因其非结构化的特性而难以被高效地检索、分析和利用。
人工观看并提炼视频信息的方法成本高昂、效率低下且无法规模化这构成了企业在智能化转型中面临的普遍瓶颈。
本项目旨在解决这一通用挑战。构建了一个由多个AI智能体Agent协作的自动化工作流该系统如同一支永不疲倦的虚拟分析师团队能够
规模化处理: 自动消化并理解海量的视频源。深度多模态理解: 同时“看懂”视频画面并“听懂”语音内容实现跨模态的信息融合。智能提炼: 将线性的、非结构化的视频信息转化为结构化的、可量化的数据资产。
以汽车行业为例一个高价值的垂域应用场景
上述挑战在那些产品复杂、竞争激烈、且高度依赖市场反馈的行业中尤为突出。为了具体展示本系统的强大能力我们将聚焦于一个典型的高价值应用场景——新款汽车的市场舆情分析。
当一家车企发布一款新车后其成败往往在最初的“黄金72小时”就已初见端倪。市场的真实声音就分散在YouTube、B站等平台上成百上千个KOL关键意见领袖的深度评-
测视频中。车企高层迫切需要知道
市场最关注我们新车的哪些核心特性在这些特性上主流观点是正面还是负面评测中大家都在拿我们的车和哪些竞品做对比我们真正的优势和短板是什么
本Jupyter Notebook将完整地、端到端地实现一个智能体系统来回答这些问题。我们将演示这个系统如何自动处理多个关于同一款新车的评测视频并通过“探索-分析-策略”的多阶段智能体协作最终生成一份包含量化数据和深度洞察的专业级Markdown舆情洞察报告。
这不仅是一个技术演示更是一个未来商业智能工作模式的缩影——将AI智能体应用于垂直领域从而在海量信息中赢得决策先机。
2. 环境准备安装所有必需的库
# 运行此单元格以安装所有依赖项
!pip install agno baidu-aip openai opencv-python-headless moviepy pydub yt-dlp3. 导入核心库
import os
import json
import base64
import re
import logging
from typing import Dict, List, Iterator
from concurrent.futures import ThreadPoolExecutor, as_completed# 框架与核心工具
from agno.agent import Agent, RunResponse
from agno.workflow import Workflow
from agno.models.openai import OpenAIChat
from aip import AipSpeech
import cv2
from moviepy.editor import VideoFileClip
from pydub import AudioSegment
import yt_dlp
from openai import OpenAI
from tqdm.notebook import tqdm
from IPython.display import display, Markdown4. 配置中心
百度智能云的语音识别相关配置需登录/申请使用,文心多模态大模型在星河社区首页点击即送~class Config:# --- 百度智能云 API 配置 ---BAIDU_APP_ID YOUR_BAIDU_APP_IDBAIDU_API_KEY YOUR_BAIDU_API_KEYBAIDU_SECRET_KEY YOUR_BAIDU_SECRET_KEY# --- 文心多模态大模型 API 配置 ---# ERNIE_MODEL_NAME可选ernie-4.5-turbo-vl或ernie-4.5-vl-28b-a3bERNIE_API_KEY YOUR_ERNIE_API_KEYERNIE_BASE_URL https://aistudio.baidu.com/llm/lmapi/v3 ERNIE_MODEL_NAME ernie-4.5-turbo-vl # --- 视频源配置 (请提供1个或多个YouTube评测视频链接) ---VIDEO_URLS [https://www.youtube.com/watch?vl-SSk-fuNl8, https://www.youtube.com/watch?vYJ_Qs-zziMc]# --- 工作目录与分析参数 ---WORKING_DIRECTORY ./auto_market_researchCHUNK_DURATION_SECONDS 30LANGUAGE_MAP {普通话: 1537, 英语: 1737, 粤语: 1637, 四川话: 1837}SELECTED_LANGUAGE 普通话# --- 并发与重试 ---MAX_CONCURRENCY 4MAX_RETRIES 3# --- 日志与环境设置 ---
os.makedirs(Config.WORKING_DIRECTORY, exist_okTrue)
logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s)
logger logging.getLogger(VideoAnalysisSystem)5. 辅助工具与预处理模块
包含为Agent工作流准备数据的所有底层函数。它是一个确定性的数据处理管道负责将视频URL转换为结构化的分析日志。
def extract_json_from_response(text: str) - dict or list:从LLM的返回文本中稳健地提取JSON对象或列表。if not text or not isinstance(text, str): return {}match re.search(r(?:json)?\s*(\{.*\}|\[.*\])\s*, text, re.DOTALL)if match: json_str match.group(1)else:match re.search(r(\{.*\}|\[.*\]), text, re.DOTALL)if not match: return {}json_str match.group(0)try:return json.loads(json_str)except json.JSONDecodeError: return {}def download_video_with_yt_dlp(video_url: str, output_dir: str, video_id: str) - str:使用 yt-dlp 下载视频限制分辨率为最高720p以加快处理速度。full_video_path os.path.join(output_dir, f{video_id}.mp4)ydl_opts {format: bestvideo[extmp4][height720]bestaudio[extm4a]/best[extmp4][height720]/best,outtmpl: full_video_path, quiet: True, noplaylist: True, retries: 3,}try:with yt_dlp.YoutubeDL(ydl_opts) as ydl:ydl.download([video_url])return full_video_path if os.path.exists(full_video_path) else Noneexcept Exception as e:logger.error(fyt-dlp 下载出错 for {video_url}: {e})return Nonedef run_audio_transcription(video_path: str, config: Config) - List[Dict]:从视频中提取音频并进行转录。temp_audio_path os.path.join(config.WORKING_DIRECTORY, ftemp_audio_{os.path.basename(video_path)}.wav)try:with VideoFileClip(video_path) as video:video.audio.write_audiofile(temp_audio_path, codecpcm_s16le, fps16000, loggerNone)full_audio AudioSegment.from_wav(temp_audio_path).set_channels(1)asr_client AipSpeech(config.BAIDU_APP_ID, config.BAIDU_API_KEY, config.BAIDU_SECRET_KEY)dev_pid config.LANGUAGE_MAP.get(config.SELECTED_LANGUAGE, 1537)tasks [{start_time: i / 1000.0, data: full_audio[i:i config.CHUNK_DURATION_SECONDS * 1000].raw_data}for i in range(0, len(full_audio), config.CHUNK_DURATION_SECONDS * 1000)if len(full_audio[i:i config.CHUNK_DURATION_SECONDS * 1000].raw_data) 1000]transcripts []with ThreadPoolExecutor(max_workersconfig.MAX_CONCURRENCY) as executor:future_to_task {executor.submit(asr_client.asr, task[data], pcm, 16000, {dev_pid: dev_pid}): task for task in tasks}for future in tqdm(as_completed(future_to_task), totallen(tasks), descf音频转录 ({os.path.basename(video_path)})):result future.result()if result and result.get(err_no) 0 and result.get(result):task_info future_to_task[future]transcripts.append({start_time: task_info[start_time], end_time: task_info[start_time] config.CHUNK_DURATION_SECONDS, text: .join(result[result])})return sorted(transcripts, keylambda x: x[start_time])finally:if os.path.exists(temp_audio_path): os.remove(temp_audio_path)def run_vision_analysis(video_path: str, transcripts: List[Dict], config: Config) - List[Dict]:对视频帧进行分块分析。vision_client OpenAI(api_keyconfig.ERNIE_API_KEY, base_urlconfig.ERNIE_BASE_URL)cap cv2.VideoCapture(video_path)duration cap.get(cv2.CAP_PROP_FRAME_COUNT) / cap.get(cv2.CAP_PROP_FPS)tasks []for start_s in range(0, int(duration), config.CHUNK_DURATION_SECONDS):end_s min(start_s config.CHUNK_DURATION_SECONDS, duration)frames_b64 []for i in range(int(end_s - start_s)):cap.set(cv2.CAP_PROP_POS_MSEC, (start_s i) * 1000)ret, frame cap.read()if ret:_, buffer cv2.imencode(.jpg, frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])frames_b64.append(base64.b64encode(buffer).decode(utf-8))if frames_b64:context_transcript .join([t[text] for t in transcripts if max(start_s, t[start_time]) min(end_s, t[end_time])])tasks.append({start_time: start_s, end_time: end_s, frames: frames_b64, transcript: context_transcript})cap.release()vision_analysis []with ThreadPoolExecutor(max_workersconfig.MAX_CONCURRENCY) as executor:future_to_task {}for task in tasks:prompt f你是一个视频分析AI。请结合语音内容“{task[transcript]}”和以下图像帧用一段话详细描述从{int(task[start_time])}秒到{int(task[end_time])}秒的视频场景和核心事件。content_list [{type: text, text: prompt}] [{type: image_url, image_url: {url: fdata:image/jpeg;base64,{f}}} for f in task[frames]]messages [{role: user, content: content_list}]future executor.submit(vision_client.chat.completions.create, modelconfig.ERNIE_MODEL_NAME, messagesmessages, temperature0.1, max_tokens2048)future_to_task[future] taskfor future in tqdm(as_completed(future_to_task), totallen(tasks), descf视觉分析 ({os.path.basename(video_path)})):response future.result()task_info future_to_task[future]vision_analysis.append({start_time: task_info[start_time], end_time: task_info[end_time], description: response.choices[0].message.content.strip()})return sorted(vision_analysis, keylambda x: x[start_time])def batch_preprocess_videos(config: Config) - Dict[str, str]:批量预处理视频并增加缓存机制跳过已处理的视频。video_log_paths {}for url in config.VIDEO_URLS:try:video_id url.split()[-1].split()[0]log_path os.path.join(config.WORKING_DIRECTORY, flog_{video_id}.json)if os.path.exists(log_path):logger.info(f✅ [缓存] 发现已处理的日志跳过预处理: {video_id})video_log_paths[video_id] log_pathcontinuelogger.info(f--- 开始预处理新视频: {video_id} ---)video_path download_video_with_yt_dlp(url, config.WORKING_DIRECTORY, video_id)if not video_path: continuetranscripts run_audio_transcription(video_path, config)vision_analysis run_vision_analysis(video_path, transcripts, config)analysis_log [{start_time: v[start_time], end_time: v[end_time], transcript: .join([t[text] for t in transcripts if max(v[start_time], t[start_time]) min(v[end_time], t[end_time])]), vision_description: v[description]} for v in vision_analysis]with open(log_path, w, encodingutf-8) as f:json.dump(analysis_log, f, indent2, ensure_asciiFalse)video_log_paths[video_id] log_pathlogger.info(f✅ 视频 {video_id} 预处理完成 - {log_path})except Exception as e:logger.error(f❌ 视频 {url} 预处理失败: {e}, exc_infoTrue)return video_log_paths6. Agent 工作流定义
这是项目的核心大脑定义了三个智能体角色探索者、分析师、策略师以及驱动它们的agno工作流。
def create_llm(config: Config, temperature: float 0.2):创建配置好的LLM实例return OpenAIChat(idconfig.ERNIE_MODEL_NAME, api_keyconfig.ERNIE_API_KEY, base_urlconfig.ERNIE_BASE_URL,temperaturetemperature)class AutoMarketResearchWorkflow(Workflow):def __init__(self, config: Config, video_log_paths: Dict[str, str]):super().__init__()self.config configself.video_logs {vid: json.load(open(path, r, encodingutf-8)) for vid, path in video_log_paths.items()}def run(self) - Iterator[RunResponse]:# --- 阶段1: 探索 Agent (自动发现关键维度) ---yield RunResponse(content--- **阶段1: 探索Agent** 正在自动发现核心分析维度 ---\n)explorer_agent Agent(modelcreate_llm(self.config, temperature0.0),instructions[你是一个市场洞察专家任务是从多个视频的分析日志中自动归纳出被反复讨论的【产品核心特性】和被明确提及的【竞争对手】。你的输出必须是一个格式正确的JSON对象包含key_features和competitors两个列表。])all_logs_text json.dumps(self.video_logs, ensure_asciiFalse, indent2)response explorer_agent.run(f请分析以下来自多个视频的日志归纳出核心特性和竞争对手。\njson\n{all_logs_text}\n)discovered_dims extract_json_from_response(response.content)self.session_state[key_features] discovered_dims.get(key_features, [])self.session_state[competitors] discovered_dims.get(competitors, [])yield RunResponse(contentf✅ **探索完成!**\n - **发现特性**: {self.session_state[key_features]}\n - **发现竞品**: {self.session_state[competitors]}\n)# --- 阶段2: 分析师 Agent (基于发现的维度进行标注) ---yield RunResponse(content\n--- **阶段2: 分析师Agent** 正在对视频进行聚焦分析和情感标注 ---\n)analyst_agent Agent(modelcreate_llm(self.config),instructions[你是一位严谨的分析师任务是精读一份视频日志抽取出所有关于指定【分析维度】的观点并标注情感。你的输出必须是一个格式正确的JSON列表每个对象包含dimension, sentiment (positive, negative, neutral), quote, 和 timestamp。])tagged_reports {}analysis_dims self.session_state[key_features] self.session_state[competitors]for video_id, log_data in self.video_logs.items():prompt f分析以下视频日志抽取出所有关于【分析维度: {analysis_dims}】的观点。\n日志:\njson\n{json.dumps(log_data, ensure_asciiFalse, indent2)}\nresponse analyst_agent.run(prompt)tagged_reports[video_id] extract_json_from_response(response.content)self.session_state[tagged_reports] tagged_reportsyield RunResponse(contentf✅ **观点标注完成**已处理所有视频。\n)# --- 阶段3: 策略师 Agent (整合数据并撰写报告) ---yield RunResponse(content\n--- **阶段3: 市场策略师Agent** 正在撰写最终洞察报告 ---\n)consolidated_db {}for video_id, report in tagged_reports.items():if isinstance(report, list):for item in report:if isinstance(item, dict) and dimension in item:dim item[dimension]if dim not in consolidated_db: consolidated_db[dim] []consolidated_db[dim].append({**item, source_video: video_id})strategist_agent Agent(modelcreate_llm(self.config, temperature0.4),instructions[你是一位高级市场策略师任务是基于一份聚合后的舆情数据库撰写一份给公司高层看的、专业的Markdown市场洞察报告。报告必须包含1. 核心特性正负面评价统计分析。2. 主要优缺点总结引用原话。3. 竞品对比分析。4. 结论与建议。])prompt f这是关于新款汽车的聚合舆情数据。请基于此数据撰写一份深入的Markdown洞察报告。\n聚合舆情数据库:\njson\n{json.dumps(consolidated_db, ensure_asciiFalse, indent2)}\nresponse strategist_agent.run(prompt)final_report response.contentself.session_state[final_report] final_reportyield RunResponse(content\n--- **工作流成功结束** ---)yield RunResponse(content\n\n *80 \n 最终舆情洞察报告\n *80 \n\n)yield RunResponse(contentfinal_report)7. 主程序执行
运行此单元格将启动整个流程并最终在下方渲染出完整的Markdown报告。
def main():主执行入口# 检查配置是否填写if not Config.BAIDU_APP_ID or YOUR_ in Config.BAIDU_APP_ID or \not Config.ERNIE_API_KEY or YOUR_ in Config.ERNIE_API_KEY:logger.error(❌ 致命错误: 请在第4步的 Config 类中填入您真实的百度智能云和文心模型的 API Keys。)return# 步骤1: 批量预处理带有缓存机制logger.info( 开始批量视频预处理...)video_log_paths batch_preprocess_videos(Config)if not video_log_paths:logger.error(所有视频均预处理失败工作流终止。请检查视频链接或网络连接。)return# 步骤2: 初始化并运行Agent工作流logger.info(\n 预处理完成开始运行Agent工作流...)workflow AutoMarketResearchWorkflow(Config, video_log_paths)final_report_content # 流式打印工作流的每一步输出for response in workflow.run():print(response.content, end)# 持续捕获最终报告的完整内容if 最终舆情洞察报告 in response.content:final_report_content # 报告开始清空内容else:final_report_content response.content# 步骤3: 使用Markdown格式优雅地展示最终报告logger.info(\n\n 工作流执行完毕正在渲染最终报告...)final_report_from_state workflow.session_state.get(final_report)if final_report_from_state:# 使用Markdown格式优雅地展示最终报告display(Markdown(final_report_from_state))else:# 如果因为某些原因 state 中没有报告再给出一个明确的失败提示error_message # 未能生成最终报告\n工作流已结束但在 session_state 中未找到 final_report。请检查策略师Agent的运行日志。display(Markdown(error_message))# --- 运行 ---
if __name__ __main__:main()最终输出新款汽车市场洞察报告
1. 核心特性正负面评价统计分析
根据聚合舆情数据库我们对新款汽车的核心特性进行了正负面评价的统计分析。以下是主要特性的情感分布
3.0T直列六缸发动机: 正面评价: 100%48伏轻混系统: 正面评价: 100%采埃孚变速箱: 正面评价: 100%前置适时四驱和多片离合器: 正面评价: 100%华为ADS 2.0智驾系统: 正面评价: 100%CDC可变阻尼减震器和单腔空气悬挂: 正面评价: 100%竞品对比奔驰、宝马、奥迪: 负面评价较多
2. 主要优缺点总结
优点3.0T直列六缸发动机:
用户评价“3.0t直接最大马力381匹最大扭矩520牛米百公里加速仅需4.8秒。”48伏轻混系统:
用户评价“车加入了48伏轻混系统当速度起来之后发动机介入动力采用了直列六缸布局可以做到动力输出更强劲更平稳。”采埃孚变速箱:
用户评价“变速箱的换挡逻辑清晰仿佛它能读懂你的心思迅速降低到合适的档位既满足了动力需求同时也尽量减少了更多的顿挫感。”前置适时四驱和多片离合器:
用户评价“在非铺装路面上它才会自动切换到四驱可以提供更好的驾驶体验和稳定性。”华为ADS 2.0智驾系统:
用户评价“它搭载了华为研发的ADS 2.0智驾系统基本上哪里都能开。”CDC可变阻尼减震器和单腔空气悬挂:
用户评价“底盘采用了电动四驱领先全系标配的CDC可变阻尼减震器和单腔空气悬挂不论是配置还是用料都属于上乘。”缺点
竞品对比中的负面评价:
用户评价奔驰、宝马、奥迪: “呆个两三年哪有BBA奔驰、宝马、奥迪有面子甚至不如奥迪A6L有面子一个小康出品的车吧就是个杂牌货被你吹得像个豪车。”3. 竞品对比分析奔驰:
负面评价集中在品牌价值方面认为新款汽车不如奔驰有面子。用户评价“你你却还停留在70、80后的BBA奔驰、宝马、奥迪时代问界M9提鞋都不配。”宝马:
正面评价宝马的驾驶体验仍然受到认可。负面评价认为新款汽车在某些方面已经超越宝马。用户评价“论舒适豪华都说奔驰舒适豪华但问界M9比奔驰还要舒适内饰要比奔驰E还豪华高档加速也比宝马快。”奥迪:
负面评价与奔驰、宝马类似认为新款汽车在某些方面已经超越奥迪。用户评价“满大街的BBA眼下国内很多款车型已经超越了我们一代人过时的梦想了。”问界M9:
正面评价在多个方面舒适性、内饰豪华度、加速性能、智能化和辅助驾驶都得到了高度评价。用户评价“买问界M9的车主那都是真大哥家里都是已经有两部三部车子的了都是家里已经有奔驰、宝马、奥迪的了再买问界M9是为了提高享受生活提升出行的品质。”4. 结论与建议
结论
新款汽车在动力性能、驾驶体验、智能化和辅助驾驶方面得到了用户的高度评价。竞品对比中奔驰、宝马和奥迪在品牌价值方面仍然有一定优势但新款汽车在多个方面已经表现出超越竞品的潜力。问界M9作为新款汽车的竞品在舒适性、内饰豪华度、加速性能、智能化和辅助驾驶方面都得到了用户的认可。
建议
品牌建设加强品牌宣传提升品牌价值和知名度以更好地与奔驰、宝马和奥迪等竞品竞争。产品优化继续优化产品性能特别是在智能化和辅助驾驶方面以保持领先地位。市场定位明确市场定位针对已有豪车如奔驰、宝马、奥迪的车主进行精准营销强调新款汽车在提升生活品质和出行体验方面的优势。
通过以上分析和建议我们相信新款汽车在市场上将有更大的发展潜力和竞争力。
