1. 视频问答智能体的技术架构解析视频问答智能体的核心目标是将任意视频转化为可交互的知识库让用户能够通过自然语言提问获取精准答案。这个系统由以下几个关键模块组成视频内容处理层负责视频元数据提取、字幕获取与分块语义理解层实现文本嵌入和向量索引构建混合检索层结合关键词搜索与语义搜索的优势智能路由层根据问题类型动态选择回答策略交互展示层提供直观的用户界面和交互体验这种分层架构设计使得系统能够高效处理从视频解析到智能问答的全流程。每个模块都可以独立优化比如更换不同的嵌入模型或调整检索算法而不影响其他模块的功能。2. 视频内容处理与结构化2.1 视频元数据提取系统首先通过YouTube的oEmbed API获取视频的基本信息def fetch_metadata(video_id: str) - dict: oembed_url fhttps://www.youtube.com/oembed?urlhttps://www.youtube.com/watch?v{video_id}formatjson try: resp requests.get(oembed_url, timeout8) if resp.status_code 200: data resp.json() return { title: data.get(title, Unknown Title), author: data.get(author_name, Unknown Channel), thumbnail_url: fhttps://img.youtube.com/vi/{video_id}/mqdefault.jpg, video_id: video_id, } except Exception: return { # 简化后的fallback逻辑 title: fVideo ({video_id}), author: Unknown, thumbnail_url: fhttps://img.youtube.com/vi/{video_id}/mqdefault.jpg, video_id: video_id, }关键点这个函数处理了多种异常情况包括API请求失败、视频不可用等确保系统在部分功能失效时仍能提供基本服务。2.2 字幕获取与分块处理获取视频字幕后系统采用滑动窗口技术将文本分割为有重叠的块def chunk_transcript(transcript: list[dict], chunk_size: int 300, overlap: int 50) - list[dict]: words_buffer [] word_timestamps [] # 将字幕转换为单词级别的时间戳 for entry in transcript: words entry[text].split() start entry[start] duration entry.get(duration, 2.0) for i, word in enumerate(words): t start (duration * i / max(len(words), 1)) words_buffer.append(word) word_timestamps.append(t) # 滑动窗口处理 chunks [] step chunk_size - overlap i 0 while i len(words_buffer): end_idx min(i chunk_size, len(words_buffer)) chunk_words words_buffer[i:end_idx] chunk_times word_timestamps[i:end_idx] chunks.append({ text: .join(chunk_words), start_time: chunk_times[0], end_time: chunk_times[-1], }) i step return chunks这种处理方式有三大优势保留上下文连续性重叠区域确保关键信息不会恰好落在分块边界时间戳精确每个单词都有对应的时间点回答可精确定位灵活可调可根据视频内容密度调整chunk_size和overlap参数3. 语义索引与混合检索系统3.1 向量嵌入与FAISS索引系统使用OpenAI的text-embedding-3-small模型生成文本嵌入并通过FAISS构建高效的向量索引def build_index(chunks: list[dict], client: OpenAI) - tuple[faiss.Index, list[dict]]: texts [c[text] for c in chunks] embeddings get_embeddings(texts, client) # 批量获取嵌入 # 归一化处理以便使用余弦相似度 norms np.linalg.norm(embeddings, axis1, keepdimsTrue) embeddings embeddings / (norms 1e-10) dim embeddings.shape[1] index faiss.IndexFlatIP(dim) # 内积归一化后的余弦相似度 index.add(embeddings) return index, chunks在实际应用中我们需要注意批量处理通过EMBED_BATCH_SIZE参数控制每次API调用的文本数量错误处理网络请求需要设置合理的超时和重试机制内存管理大型视频可能需要分批次构建索引3.2 BM25关键词索引作为向量搜索的补充系统实现了基于BM25算法的关键词检索class KeywordIndex: def __init__(self, chunks: List[Dict]): self.chunks chunks self.corpus [chunk[text] for chunk in chunks] self.tokenized_corpus [self._tokenize(text) for text in self.corpus] self.bm25 BM25Okapi(self.tokenized_corpus) staticmethod def _tokenize(text: str) - List[str]: return re.findall(r\w, text.lower()) # 简单分词转小写、分割单词 def search(self, query: str, top_k: int 5) - List[Dict]: query_tokens self._tokenize(query) scores self.bm25.get_scores(query_tokens) top_indices sorted(range(len(scores)), keylambda i: scores[i], reverseTrue)[:top_k] return [self.chunks[idx].copy() for idx in top_indices]BM25的优势在于对精确术语匹配更敏感不依赖预训练模型计算速度快可解释性强便于调试3.3 混合检索与RRF融合系统采用倒数排名融合(Reciprocal Rank Fusion)算法结合两种检索结果def reciprocal_rank_fusion(keyword_results, vector_results, k60): # 构建排名映射 keyword_ranks { (c[start_time], c[end_time]): (rank, c) for rank, c in enumerate(keyword_results) } vector_ranks { (c[start_time], c[end_time]): (rank, c) for rank, c in enumerate(vector_results) } # 计算RRF分数 rrf_scores {} for chunk_id, (rank, _) in keyword_ranks.items(): rrf_scores[chunk_id] 1.0 / (rank k) for chunk_id, (rank, _) in vector_ranks.items(): score 1.0 / (rank k) rrf_scores[chunk_id] rrf_scores.get(chunk_id, 0) score # 合并结果并排序 all_chunks { **{cid: c for cid, (_, c) in keyword_ranks.items()}, **{cid: c for cid, (_, c) in vector_ranks.items()} } sorted_chunks sorted(all_chunks.items(), keylambda item: rrf_scores[item[0]], reverseTrue) return [chunk for _, chunk in sorted_chunks]RRF的核心思想是每个检索方法独立返回排序结果对每个结果的排名取倒数并加权求和参数k控制排名的影响程度典型值60这种融合方式既保留了语义搜索的上下文理解能力又发挥了关键词搜索的精确匹配优势。4. 智能问答与路由机制4.1 问题分类路由系统首先通过轻量级分类器判断问题类型ROUTER_PROMPT 你是一个YouTube视频问答助手的查询分类器。 将用户问题分类为两种类型 global — 需要理解整个视频的问题 如总结、概述、主要话题、关键要点、章节结构等 rag — 关于视频中特定事实、时刻、人物或概念的问题 如X何时发生、演讲者关于Y说了什么、解释概念Z等 只回复JSON对象{route: global} 或 {route: rag} 不要解释。不要其他文本。 def classify_query(user_message: str, client: OpenAI) - str: response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: ROUTER_PROMPT}, {role: user, content: user_message}, ], temperature0, max_tokens20, ) try: return json.loads(response.choices[0].message.content)[route] except: return rag # 默认安全回退这种设计带来了两个好处资源优化全局性问题不需要执行耗时的检索操作答案质量不同类型的问题采用不同的回答策略4.2 回答生成策略根据问题类型系统采用不同的提示模板全局性问题提示模板你是一个智能视频助手。你已获得带有时间戳的YouTube视频完整字幕。 要求 - 使用完整字幕全面回答用户问题 - 对于总结涵盖所有主要部分而不仅是开头 - 对于主要部分/主题识别明显的话题转换并列出每个话题及其开始时间戳 - 使用精确的Markdown格式内联引用时间戳[MM:SS](https://youtube.com/watch?v{video_id}tXs) - 结构清晰 — 使用编号列表或明确的小节特定性问题提示模板你是一个智能视频助手。你已获得YouTube视频字幕的相关摘录。 要求 - 仅基于提供的摘录回答问题 - 在答案中内联引用1-3个时间戳使用精确的Markdown格式 - 只有当时间戳直接支持你写的句子时才引用 - 不要将所有时间戳列在最后 — 自然地将其编织到答案中 - 如果上下文中没有答案说我在视频中找不到相关信息实际应用中我们发现几个关键点严格的时间戳格式要求确保生成的链接可点击明确的指令减少模型臆造答案的可能性结构化输出提升用户体验5. Streamlit交互界面实现5.1 界面布局设计系统采用经典的两栏布局# 主布局 st.set_page_config( page_titleYT Chat, layoutwide, initial_sidebar_statecollapsed, ) # 顶部导航栏 st.markdown( div classtopbar div classyt-logo▶ spanYT/span Chat/div div classurl-input-wrap !-- URL输入框 -- /div /div , unsafe_allow_htmlTrue) # 两栏主界面 left_col, right_col st.columns([1.15, 0.85], gapsmall) with left_col: # 视频嵌入区 st.markdown(f div classvideo-embed-wrap iframe srchttps://www.youtube.com/embed/{video_id}?start{start_time} frameborder0 allowfullscreen/iframe /div , unsafe_allow_htmlTrue) with right_col: # 聊天区 st.markdown( div classchat-panel div classchat-messages !-- 消息历史 -- /div div classchat-input-area !-- 输入框 -- /div /div , unsafe_allow_htmlTrue)5.2 关键交互功能时间戳跳转通过URL参数实现视频定位# 处理跳转请求 if jump_to in st.query_params: try: jump_seconds int(st.query_params[jump_to]) # 更新iframe的start参数 st.session_state.jump_to_seconds jump_seconds # 清除参数避免重复触发 st.query_params.clear() except: pass # 在视频嵌入中使用跳转时间 start_time st.session_state.get(jump_to_seconds, 0)会话状态管理维护多视频的独立聊天历史def init_state(): if videos not in st.session_state: st.session_state.videos {} # 存储视频数据 if conversations not in st.session_state: st.session_state.conversations {} # 各视频的聊天历史 if active_video_id not in st.session_state: st.session_state.active_video_id None # 获取当前活跃对话 def active_conversation(): vid st.session_state.active_video_id if vid and vid not in st.session_state.conversations: st.session_state.conversations[vid] [] return st.session_state.conversations.get(vid, [])5.3 样式优化技巧系统通过自定义CSS实现了几个重要效果时间戳按钮悬浮菜单.ts-dropdown:hover .ts-menu { display: block; animation: fadeIn 0.2s; } keyframes fadeIn { from { opacity: 0; transform: translateY(5px); } to { opacity: 1; transform: translateY(0); } }消息气泡差异化样式.msg-user { background: #2d2d2d; border-radius: 18px 18px 4px 18px; align-self: flex-end; } .msg-ai { background: transparent; border: 1px solid #3d3d3d; border-radius: 4px 18px 18px 18px; align-self: flex-start; }加载动画keyframes thinking-pulse { 0%, 80%, 100% { opacity: 0.2; transform: scale(0.8); } 40% { opacity: 1; transform: scale(1.1); } } .thinking-dot { animation: thinking-pulse 1.2s ease-in-out infinite; animation-delay: calc(var(--order) * 0.2s); }6. 部署与优化实践6.1 性能优化策略在处理长视频时我们采用了几个优化措施渐进式加载先显示已处理的部分内容后台继续处理剩余部分def process_video(video_id): # 先快速加载元数据和原始字幕 meta fetch_metadata(video_id) raw_transcript fetch_transcript(video_id) yield {stage: metadata, data: meta} # 分批次处理字幕块 batch_size 100 for i in range(0, len(raw_transcript), batch_size): batch raw_transcript[i:ibatch_size] chunks chunk_transcript(batch) yield {stage: chunks, count: len(chunks)} # 分批构建索引 if i 0: index, all_chunks build_index(chunks, client) else: texts [c[text] for c in chunks] embeddings get_embeddings(texts, client) norms np.linalg.norm(embeddings, axis1, keepdimsTrue) embeddings embeddings / (norms 1e-10) index.add(embeddings) all_chunks.extend(chunks) yield {stage: complete, index: index, chunks: all_chunks}缓存机制将处理过的视频数据持久化存储避免重复处理资源监控对大视频进行预警并限制最大处理时长6.2 错误处理与健壮性系统在多个层面实现了错误恢复机制字幕获取def fetch_transcript(video_id: str) - list[dict]: try: # 首选英语字幕 fetched YouTubeTranscriptApi().fetch(video_id, languages[en]) return [{text: s.text, start: s.start} for s in fetched] except TranscriptsDisabled: # 尝试其他可用语言 transcript_list YouTubeTranscriptApi().list(video_id) available [t.language_code for t in transcript_list] if available: fetched YouTubeTranscriptApi().fetch(video_id, languagesavailable) return [{text: s.text, start: s.start} for s in fetched] raise ValueError(No transcripts available for this video)API调用def safe_embedding(texts: list[str], client: OpenAI, max_retries3) - np.ndarray: for attempt in range(max_retries): try: return get_embeddings(texts, client) except Exception as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避6.3 实际部署注意事项API密钥管理通过环境变量注入避免硬编码from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY)资源限制对单个视频的处理设置超时和内存限制用户体验在处理过程中提供清晰的进度反馈with st.status(Processing video..., expandedTrue) as status: st.write(Fetching metadata...) meta fetch_metadata(video_id) st.write(Downloading transcript...) transcript fetch_transcript(video_id) st.write(Building search index...) index, chunks build_index(transcript, client) status.update(labelProcessing complete!, statecomplete)7. 扩展与进阶应用7.1 多模态扩展当前系统主要处理视频的字幕文本可以扩展为真正的多模态系统视觉问答使用CLIP等模型分析视频关键帧import clip model, preprocess clip.load(ViT-B/32) frames extract_key_frames(video_path) image_features model.encode_image(preprocess(frames))音频分析处理语音语调、背景音乐等非文本信息7.2 高级检索技术分层索引对长视频建立多级索引结构顶层视频章节摘要中层主题段落底层详细字幕块查询扩展使用LLM重写和扩展用户查询def expand_query(query: str, client: OpenAI) - list[str]: prompt fOriginal query: {query} Generate 3 alternative phrasings that capture the same intent but may retrieve different relevant passages. response client.chat.completions.create(...) return parse_alternatives(response)7.3 企业级应用场景培训视频知识库员工可以快速查找产品培训视频中的特定内容会议记录分析自动转录会议视频并生成可搜索的知识库教育内容检索学生通过自然语言查找课程视频中的相关讲解我在实际部署中发现系统性能与视频长度和内容密度密切相关。对于1小时左右的讲座视频最佳chunk_size在250-350词之间overlap约50词。而短视频(10分钟内)可以使用较小的chunk_size(150-200)获得更精确的时间定位。