实用提示：停止观看YouTube视频，开始构建AI智能体。
《实用笔记》操作指南：停止观看YouTube视频。立即构建AI智能体——为采用该模式的团队提供合同、校验机制以及可直接使用的代码模板。
以下笔记为“停止观看YouTube视频，构建AI智能体并与之聊天”这一目标提供了可行的实施路径。重点在于合同定义、校验机制以及可直接插入的代码占位符，而非激励性表述。 在完成概览阶段时，首先需明确合同细节：所需输入、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。
引言
在将引入阶段视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。
该项目的功能
将“该项目的作用”这一阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任主体，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在流程中断后导致无法继续执行。
架构概览
将“架构概览”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份标准范本、一个故障案例以及回滚说明。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。 保持图结构扁平且类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。 将“架构概览”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份标准范本、一个故障案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放于一处，以便操作人员无需查看整个图结构即可进行审计。
核心组件
在核心组件阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
设置：
在设置阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 对于涉及资金支出或修改生产数据的环节，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
python3 -m venv .venv
source .venv/bin/activate
python -m venv .venv
.venv\Scripts\activate
streamlit>=1.35.0
openai>=1.30.0
python-dotenv>=1.0.0
youtube-transcript-api>=0.6.2
faiss-cpu>=1.8.0
numpy>=1.26.0
requests>=2.31.0
rank-bm25>=0.2.2
#terminal
pip install requirements.txt
步骤1：获取转录文本并分块
在第一步“获取文字记录”阶段，应在修改代码之前明确输入参数、该阶段的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该阶段，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，杜绝无声的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
1.1 提取YouTube视频ID
在1.1阶段中，修改代码之前需先明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于会产生支出或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
def extract_video_id(url: str) -> str | None:
patterns = [
r"(?:v=|\/)([0-9A-Za-z_-]{11}).*",
r"(?:youtu\.be\/)([0-9A-Za-z_-]{11})",
r"(?:embed\/)([0-9A-Za-z_-]{11})",
r"^([0-9A-Za-z_-]{11})quot;,
]
for pattern in patterns:
match = re.search(pattern, url)
if match:
return match.group(1)
return None
1.2：获取转录内容
在1.2阶段获取相关信息时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个流程即可进行审核。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。
fetched = api.fetch(video_id, languages=['en'])
[{"text": …, "start": …, "duration": …}, …]
1.3：为转录内容添加时间戳并分块处理
在“1 3 分块处理”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
def chunk_transcript(transcript: list[dict], chunk_size: int = 300, overlap: int = 50) -> list[dict]:
chunks = []
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)
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({
"chunk_id": chunk_id,
"text": " ".join(chunk_words),
"start_time": chunk_times[0],
"end_time": chunk_times[-1],
})
i += step
"""
transcript.py - Fetch and parse YouTube transcripts with timestamps
Compatible with youtube-transcript-api v1.x
"""
from youtube_transcript_api import YouTubeTranscriptApi
from youtube_transcript_api._errors import (
NoTranscriptFound, TranscriptsDisabled, VideoUnavailable,
CouldNotRetrieveTranscript,
)
import re
def extract_video_id(url: str) -> str | None:
"""Extract video ID from various YouTube URL formats."""
patterns = [
r"(?:v=|\/)([0-9A-Za-z_-]{11}).*",
r"(?:youtu\.be\/)([0-9A-Za-z_-]{11})",
r"(?:embed\/)([0-9A-Za-z_-]{11})",
r"^([0-9A-Za-z_-]{11})quot;,
]
for pattern in patterns:
match = re.search(pattern, url)
if match:
return match.group(1)
return None
def fetch_transcript(video_id: str) -> list[dict]:
"""
Fetch transcript for a YouTube video.
Returns list of {text, start, duration} dicts.
Compatible with youtube-transcript-api v1.x (instance-based API).
"""
api = YouTubeTranscriptApi()
# Try English first
try:
fetched = api.fetch(video_id, languages=['en'])
return [{"text": s.text, "start": s.start, "duration": s.duration} for s in fetched]
except TranscriptsDisabled:
raise ValueError("Transcripts are disabled for this video.")
except VideoUnavailable:
raise ValueError("Video is unavailable or private.")
except (NoTranscriptFound, CouldNotRetrieveTranscript):
pass # Will try other languages below
except Exception:
pass # Will try other languages below
# Fallback: discover all available languages, use the first one
try:
transcript_list = api.list(video_id)
available = [t.language_code for t in transcript_list]
if not available:
raise ValueError("No transcripts available for this video.")
fetched = api.fetch(video_id, languages=available)
return [{"text": s.text, "start": s.start, "duration": s.duration} for s in fetched]
except TranscriptsDisabled:
raise ValueError("Transcripts are disabled for this video.")
except VideoUnavailable:
raise ValueError("Video is unavailable or private.")
except ValueError:
raise
except Exception as e:
raise ValueError(f"Could not fetch transcript: {str(e)}")
def chunk_transcript(transcript: list[dict], chunk_size: int = 300, overlap: int = 50) -> list[dict]:
"""
Chunk transcript into overlapping windows, preserving timestamps.
Each chunk: {text, start_time, end_time, chunk_id}
"""
chunks = []
words_buffer = []
word_timestamps = []
# Flatten transcript into word-level with 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)
# Slide window
step = chunk_size - overlap
chunk_id = 0
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]
chunk_text = " ".join(chunk_words)
start_time = chunk_times[0]
end_time = chunk_times[-1]
chunks.append({
"chunk_id": chunk_id,
"text": chunk_text,
"start_time": start_time,
"end_time": end_time,
})
chunk_id += 1
i += step
if end_idx == len(words_buffer):
break
return chunks
def format_timestamp(seconds: float) -> str:
"""Convert seconds to MM:SS or HH:MM:SS string."""
seconds = int(seconds)
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
def make_youtube_link(video_id: str, seconds: float) -> str:
"""Create a deep-link YouTube URL at a specific timestamp."""
t = int(seconds)
return f"https://www.youtube.com/watch?v={video_id}&t={t}s"
"""
metadata.py - Fetch YouTube video metadata (title, thumbnail, duration, channel)
"""
import requests
import json
import re
def fetch_metadata(video_id: str) -> dict:
"""
Fetch video metadata using YouTube oEmbed API + noembed fallback.
Returns dict with title, author, thumbnail_url, duration_str.
"""
# Try YouTube oEmbed (no API key needed)
oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
try:
resp = requests.get(oembed_url, timeout=8)
if resp.status_code == 200:
data = resp.json()
thumbnail = f"https://img.youtube.com/vi/{video_id}/mqdefault.jpg"
return {
"title": data.get("title", "Unknown Title"),
"author": data.get("author_name", "Unknown Channel"),
"thumbnail_url": thumbnail,
"video_id": video_id,
"url": f"https://www.youtube.com/watch?v={video_id}",
}
except Exception:
pass
# Fallback: minimal info
return {
"title": f"Video ({video_id})",
"author": "Unknown",
"thumbnail_url": f"https://img.youtube.com/vi/{video_id}/mqdefault.jpg",
"video_id": video_id,
"url": f"https://www.youtube.com/watch?v={video_id}",
}
步骤 2：嵌入转录内容
在第二步的嵌入阶段，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。必须引用实际作为答案依据的段落；没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失所致。
2.1: 嵌入与FAISS索引
在处理2 1嵌入阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 需引用实际作为答案依据的段落。没有引用的话，操作员就无法区分幻觉内容与索引缺失的问题。 在处理2 1嵌入阶段时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个操作员能够审核的位置，无需阅读整个系统结构。
embeddings = get_embeddings(texts, client)
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
embeddings = embeddings / (norms + 1e-10)
index = faiss.IndexFlatIP(dim)
index.add(embeddings)
def search_index(query, index, chunks, client, top_k=5):
response = client.embeddings.create(model=EMBEDDING_MODEL, input=[query])
q_emb = np.array([response.data[0].embedding], dtype=np.float32)
q_emb = q_emb / (np.linalg.norm(q_emb) + 1e-10)
scores, indices = index.search(q_emb, top_k)
2.2：使用BM25构建关键词索引
在处理“构建关键词索引”这一阶段时，首先明确相关要求：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会偏离原定方向。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都属于产品功能的一部分，而非后续需要补充的内容。 在成本较高的操作之后设置检查点。当操作员重新尝试某个节点时，恢复流程不应再次调用相同的LLM接口。
self.tokenized_corpus = [self._tokenize(text) for text in self.corpus]
self.bm25 = BM25Okapi(self.tokenized_corpus)
query_tokens = self._tokenize(query)
scores = self.bm25.get_scores(query_tokens)
"""
embedder.py - Embed transcript chunks and build a FAISS index
"""
import numpy as np
import faiss
import pickle
import os
from openai import OpenAI
EMBEDDING_MODEL = "text-embedding-3-small"
EMBED_BATCH_SIZE = 64
def get_embeddings(texts: list[str], client: OpenAI) -> np.ndarray:
"""Embed a list of texts using OpenAI embeddings in batches."""
all_embeddings = []
for i in range(0, len(texts), EMBED_BATCH_SIZE):
batch = texts[i:i + EMBED_BATCH_SIZE]
response = client.embeddings.create(model=EMBEDDING_MODEL, input=batch)
batch_embeddings = [item.embedding for item in response.data]
all_embeddings.extend(batch_embeddings)
return np.array(all_embeddings, dtype=np.float32)
def build_index(chunks: list[dict], client: OpenAI) -> tuple[faiss.Index, list[dict]]:
"""
Build a FAISS flat L2 index from transcript chunks.
Returns (index, chunks) — chunks are stored as metadata alongside index.
"""
texts = [c["text"] for c in chunks]
embeddings = get_embeddings(texts, client)
# Normalize for cosine similarity
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
embeddings = embeddings / (norms + 1e-10)
dim = embeddings.shape[1]
index = faiss.IndexFlatIP(dim) # Inner product = cosine after normalization
index.add(embeddings)
return index, chunks
def search_index(
query: str,
index: faiss.Index,
chunks: list[dict],
client: OpenAI,
top_k: int = 5,
) -> list[dict]:
"""
Search the FAISS index for chunks most relevant to query.
Returns top_k chunks with similarity scores.
"""
response = client.embeddings.create(model=EMBEDDING_MODEL, input=[query])
q_emb = np.array([response.data[0].embedding], dtype=np.float32)
# Normalize
q_emb = q_emb / (np.linalg.norm(q_emb) + 1e-10)
scores, indices = index.search(q_emb, top_k)
results = []
for score, idx in zip(scores[0], indices[0]):
if idx < len(chunks):
chunk = chunks[idx].copy()
chunk["score"] = float(score)
results.append(chunk)
return results
"""
keyword_index.py — BM25 keyword indexing and retrieval for transcript chunks.
Provides efficient keyword-based search as complement to semantic vector search.
"""
from rank_bm25 import BM25Okapi
from typing import List, Dict
class KeywordIndex:
"""Build and search a BM25 keyword index from transcript chunks."""
def __init__(self, chunks: List[Dict]):
"""
Initialize BM25 index from chunks.
Args:
chunks: List of chunk dicts with 'text', 'start_time', 'end_time' keys
"""
self.chunks = chunks
self.corpus = [chunk["text"] for chunk in chunks]
# Tokenize: split on whitespace, lowercase, simple punctuation removal
self.tokenized_corpus = [self._tokenize(text) for text in self.corpus]
self.bm25 = BM25Okapi(self.tokenized_corpus)
@staticmethod
def _tokenize(text: str) -> List[str]:
"""Simple tokenization: lowercase, split on whitespace."""
import re
# Convert to lowercase, split on whitespace, remove punctuation
tokens = re.findall(r'\w+', text.lower())
return tokens
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""
Search BM25 index for chunks matching query.
Args:
query: Search query string
top_k: Number of top results to return
Returns:
List of chunks with 'bm25_score' added; sorted by score descending
"""
query_tokens = self._tokenize(query)
# BM25 returns scores for each document in corpus
scores = self.bm25.get_scores(query_tokens)
# Sort by score descending, get top_k indices
top_indices = sorted(
range(len(scores)),
key=lambda i: scores[i],
reverse=True
)[:top_k]
results = []
for idx in top_indices:
if idx < len(self.chunks):
chunk = self.chunks[idx].copy()
chunk["bm25_score"] = float(scores[idx])
results.append(chunk)
return results
def build_keyword_index(chunks: List[Dict]) -> KeywordIndex:
"""
Convenience function to build a BM25 index from chunks.
Args:
chunks: List of chunk dicts
Returns:
KeywordIndex instance ready for search
"""
return KeywordIndex(chunks)
步骤3：结合RRF融合的混合检索方式
在处理第3步的混合检索阶段时，首先写下相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非冗长的脚本。当某一步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难解决检索效果不佳的问题。
score = sum(1 / (rank + k))
第4步：查询路由——全局与特定
在处理第4步查询路由阶段时，首先需明确相关契约：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关组件命名，定义成功检测标准，并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的LLM接口。 在处理第4步查询路由阶段时，首先需明确相关契约：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储及功能开关应集中存放于一个位置，以便操作员无需查看整个架构即可进行审计。
response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": user_message},
],
)
"""
retrieval_fusion.py — Reciprocal Rank Fusion (RRF) for hybrid search.
Merges keyword (BM25) and semantic (vector) results into a single ranked list.
Uses reciprocal rank fusion formula: score = sum(1 / (rank + k))
"""
from typing import List, Dict, Tuple
def reciprocal_rank_fusion(
keyword_results: List[Dict],
vector_results: List[Dict],
k: int = 60,
) -> List[Dict]:
"""
Merge and rank results from keyword and vector searches using RRF.
RRF formula for each result:
score = sum(1 / (rank_keyword + k) + 1 / (rank_vector + k))
Where rank is 0-indexed position in each result list.
Results appearing in both lists get scores from both; results in one list only
contribute their single score.
Args:
keyword_results: List of chunks from BM25 search (with 'bm25_score' field)
vector_results: List of chunks from vector search (with 'score' field for cosine similarity)
k: RRF parameter; higher k diminishes effect of rank position
Default 60 is standard; tune based on result quality
Returns:
List of unique chunks sorted by fused RRF score (descending)
Each chunk has 'rrf_score', 'keyword_rank', 'vector_rank' fields added
"""
# Build ranking maps: chunk_id -> (rank, original_chunk_dict)
# Using (start_time, end_time) as unique chunk ID
keyword_ranks: Dict[Tuple, Tuple[int, Dict]] = {}
vector_ranks: Dict[Tuple, Tuple[int, Dict]] = {}
for rank, chunk in enumerate(keyword_results):
chunk_id = (chunk.get("start_time"), chunk.get("end_time"))
keyword_ranks[chunk_id] = (rank, chunk)
for rank, chunk in enumerate(vector_results):
chunk_id = (chunk.get("start_time"), chunk.get("end_time"))
vector_ranks[chunk_id] = (rank, chunk)
# Compute RRF scores for all unique chunks
rrf_scores: Dict[Tuple, float] = {}
all_chunks: Dict[Tuple, Dict] = {}
chunk_metadata: Dict[Tuple, Dict] = {} # Track rank info
# Process keyword results
for chunk_id, (rank, chunk) in keyword_ranks.items():
rrf_scores[chunk_id] = 1.0 / (rank + k)
all_chunks[chunk_id] = chunk
chunk_metadata[chunk_id] = {"keyword_rank": rank, "vector_rank": None}
# Process vector results
for chunk_id, (rank, chunk) in vector_ranks.items():
vector_contribution = 1.0 / (rank + k)
if chunk_id in rrf_scores:
rrf_scores[chunk_id] += vector_contribution
chunk_metadata[chunk_id]["vector_rank"] = rank
else:
rrf_scores[chunk_id] = vector_contribution
all_chunks[chunk_id] = chunk
chunk_metadata[chunk_id] = {"keyword_rank": None, "vector_rank": rank}
# Sort by RRF score descending
sorted_chunks = sorted(
all_chunks.items(),
key=lambda item: rrf_scores[item[0]],
reverse=True
)
# Build result list with metadata
results = []
for chunk_id, chunk in sorted_chunks:
result_chunk = chunk.copy()
result_chunk["rrf_score"] = rrf_scores[chunk_id]
result_chunk["keyword_rank"] = chunk_metadata[chunk_id]["keyword_rank"]
result_chunk["vector_rank"] = chunk_metadata[chunk_id]["vector_rank"]
results.append(result_chunk)
return results
def fuse_and_get_top_k(
keyword_results: List[Dict],
vector_results: List[Dict],
top_k: int = 5,
rrf_k: int = 60,
) -> List[Dict]:
"""
Convenience function: fuse results and return top_k.
Args:
keyword_results: BM25 search results
vector_results: Vector search results
top_k: Number of results to return from fused list
rrf_k: RRF parameter
Returns:
Top k chunks from fused ranking
"""
fused = reciprocal_rank_fusion(keyword_results, vector_results, k=rrf_k)
return fused[:top_k]
第6步：构建提示词
将第6步的构建过程视为可度量的工作面效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 同时记录成功路径和恢复路径。重试机制、人工审核环节以及错误处理都是产品的一部分，而非后续需要补充的内容。 为每轮对话和每次会话设定token预算。智能工具会大量消耗上下文信息，设置上限可避免演示过程变成意外的费用账单。
[MM:SS](https://www.youtube.com/watch?v={video_id}&t=Xs)
"""
chat.py — LLM-routed chat with inline timestamp citations.
Router: one fast GPT call → "global" | "rag"
Global: full transcript passed as context (up to 80k tokens)
RAG: hybrid retrieval (BM25 keyword + vector semantic) with RRF fusion,
LLM cites 1-2 timestamps inline in answer
"""
from openai import OpenAI
from embedder import search_index
from keyword_index import build_keyword_index
from retrieval_fusion import fuse_and_get_top_k
from transcript import format_timestamp, make_youtube_link
import faiss
import json
MODEL = "gpt-4o-mini"
MAX_FULL_TRANSCRIPT_WORDS = 60_000
# ── System prompts ─────────────────────────────────────────────────────────────
ROUTER_PROMPT = """You are a query classifier for a YouTube video Q&A assistant.
Classify the user's question as one of two types:
"global" — The question requires understanding the ENTIRE video.
Examples: summarize, overview, main topics, key takeaways,
chapters, structure, what is this video about, full recap.
"rag" — The question is about a SPECIFIC fact, moment, person, concept,
or timestamp in the video. Examples: when did X happen,
what did the speaker say about Y, explain concept Z.
Reply with ONLY a JSON object: {"route": "global"} or {"route": "rag"}
No explanation. No other text."""
GLOBAL_SYSTEM_PROMPT = """You are an intelligent video assistant. You have been given the COMPLETE transcript of a YouTube video with timestamps.
Instructions:
- Answer the user's question using the full transcript comprehensively.
- For summaries: cover ALL major sections, not just the beginning.
- For "main sections/topics": identify distinct topic shifts and list each with its start timestamp.
- Cite timestamps inline using this EXACT markdown format: [MM:SS](https://www.youtube.com/watch?v={video_id}&t=Xs)
where X is the timestamp in seconds. Always include the 's' suffix after the number. Example: &t=315s not &t=315
- Be well-structured — use numbered lists or clear sections.
- Only cite timestamps that are genuinely relevant to that point.
"""
SPECIFIC_SYSTEM_PROMPT = """You are an intelligent video assistant. You have been given relevant excerpts from a YouTube video transcript.
Instructions:
- Answer the question based ONLY on the provided transcript excerpts.
- Cite 1 to 3 timestamps INLINE in your answer using this EXACT markdown format:
[MM:SS](https://www.youtube.com/watch?v={video_id}&t=Xs)
where X is the timestamp in seconds. Always include the 's' suffix. Example: &t=315s not &t=315
- Only cite a timestamp when it directly supports the specific sentence you are writing.
- Do NOT list all timestamps at the end — weave them naturally into the answer.
- If the answer spans multiple parts of the video, show each part as a numbered point with its own inline timestamp.
- If the context does not contain the answer, say: "I couldn't find information about that in this video."
- Never make up information not present in the provided context.
"""
# ── LLM Router ────────────────────────────────────────────────────────────────
def classify_query(user_message: str, client: OpenAI) -> str:
"""
Ask GPT-4o-mini to classify the query as 'global' or 'rag'.
Falls back to 'rag' on any error.
"""
try:
response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": ROUTER_PROMPT},
{"role": "user", "content": user_message},
],
temperature=0,
max_tokens=20,
)
raw = response.choices[0].message.content.strip()
parsed = json.loads(raw)
route = parsed.get("route", "rag")
return route if route in ("global", "rag") else "rag"
except Exception:
return "rag" # safe default
# ── Context builders ──────────────────────────────────────────────────────────
def build_full_transcript_context(all_chunks: list[dict]) -> str:
"""Concatenate ALL chunks sorted by time, capped at MAX_FULL_TRANSCRIPT_WORDS."""
sorted_chunks = sorted(all_chunks, key=lambda c: c["start_time"])
parts = []
words_so_far = 0
for chunk in sorted_chunks:
chunk_words = len(chunk["text"].split())
if words_so_far + chunk_words > MAX_FULL_TRANSCRIPT_WORDS:
parts.append("[... transcript truncated for length ...]")
break
ts = format_timestamp(chunk["start_time"])
parts.append(f"[{ts}] {chunk['text']}")
words_so_far += chunk_words
return "\n".join(parts)
def build_rag_context(chunks: list[dict]) -> str:
"""Format retrieved chunks into a timestamped context block."""
parts = []
for chunk in chunks:
ts = format_timestamp(chunk["start_time"])
end_ts = format_timestamp(chunk["end_time"])
parts.append(f"[{ts} - {end_ts}]\n{chunk['text']}")
return "\n\n---\n\n".join(parts)
# ── Source extractor (parses inline links from LLM reply) ────────────────────
def extract_sources_from_reply(reply: str, video_id: str) -> list[dict]:
"""
Parse timestamp markdown links that the LLM wrote inline.
Matches patterns like [4:32](https://...&t=272s)
Returns deduplicated list of {timestamp, seconds, link}.
"""
import re
# Match [MM:SS] or [H:MM:SS] followed by a YouTube URL with &t=Xs
# s suffix is optional — LLM sometimes writes &t=315 not &t=315s
pattern = r'\[([\d]{1,2}:\d{2}(?::\d{2})?)\]\((https://www\.youtube\.com/watch\?v=[\w-]+&t=(\d+)s?)\)'
matches = re.findall(pattern, reply)
seen = set()
sources = []
for ts_label, url, seconds_str in matches:
seconds = int(seconds_str)
if seconds not in seen:
seen.add(seconds)
sources.append({
"timestamp": ts_label,
"seconds": float(seconds),
"link": url,
})
return sources
# ── Main chat function ────────────────────────────────────────────────────────
def chat_with_video(
user_message: str,
conversation_history: list[dict],
index: faiss.Index,
chunks: list[dict],
video_id: str,
client: OpenAI,
top_k: int = 5,
keyword_index=None,
) -> tuple[str, list[dict]]:
"""
1. Classify query → global | rag
2. Build context accordingly
3. For 'rag': use hybrid retrieval (BM25 + semantic with RRF fusion)
4. Call GPT-4o-mini with inline-timestamp instructions
5. Parse timestamps from reply for UI chips
Args:
user_message: User's query
conversation_history: Previous messages in conversation
index: FAISS vector index
chunks: All transcript chunks
video_id: YouTube video ID
client: OpenAI client
top_k: Number of results to return after fusion
keyword_index: KeywordIndex instance for BM25 search (optional)
Returns:
Tuple of (reply_text, sources)
"""
# ── Step 1: Route ──────────────────────────────────────────────────────
route = classify_query(user_message, client)
# ── Step 2: Build context ──────────────────────────────────────────────
if route == "global":
context = build_full_transcript_context(chunks)
system_prompt = GLOBAL_SYSTEM_PROMPT.replace("{video_id}", video_id)
context_label = "FULL VIDEO TRANSCRIPT (with timestamps):"
max_tokens = 1800
else:
# ── Hybrid retrieval: BM25 keyword + semantic vector with RRF fusion ──
# Retrieve top 10 from each method, then fuse to top_k
vector_results = search_index(user_message, index, chunks, client, top_k=10)
if keyword_index is not None:
keyword_results = keyword_index.search(user_message, top_k=10)
# Fuse using Reciprocal Rank Fusion
retrieved = fuse_and_get_top_k(
keyword_results,
vector_results,
top_k=top_k,
rrf_k=60
)
else:
# Fallback: use vector search only if keyword index not available
retrieved = vector_results[:top_k]
context = build_rag_context(retrieved)
system_prompt = SPECIFIC_SYSTEM_PROMPT.replace("{video_id}", video_id)
context_label = "RELEVANT TRANSCRIPT EXCERPTS (hybrid keyword + semantic search):"
max_tokens = 900
# ── Step 3: Call LLM ───────────────────────────────────────────────────
messages = [
{"role": "system", "content": system_prompt},
{"role": "system", "content": f"{context_label}\n\n{context}"},
]
messages.extend(conversation_history)
messages.append({"role": "user", "content": user_message})
response = client.chat.completions.create(
model=MODEL,
messages=messages,
temperature=0,
max_tokens=max_tokens,
)
reply = response.choices[0].message.content
# ── Step 4: Extract inline timestamp links as UI chips ─────────────────
sources = extract_sources_from_reply(reply, video_id)
return reply, sources
第7步：提取来源时间戳
将第7步“提取数据源阶段”视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出现故障时，故障应能指向单一责任主体，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在流程中断后导致无法继续执行。
pattern = r'\[([\d]{1,2}:\d{2}(?::\d{2})?)\]\((https://www\.youtube\.com/watch\?v=[\w-]+&t=(\d+)s?)\)'
第8步：Streamlit用户界面与会话状态
将 Step 8 的 Streamlit UI 阶段视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想运行案例、一个故障案例以及回滚说明。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，杜绝默许的半完成状态。 保持图表状态简洁且具有类型约束。嵌套的数据块会掩盖具体是哪个节点修改了哪个字段，还会在中断后导致无法继续处理。 将 Step 8 的 Streamlit UI 阶段视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想运行案例、一个故障案例以及回滚说明。 将配置信息置于应用程序代码之外。环境文件、密钥存储及功能开关应集中存放于一处，以便操作人员无需查看整个图表即可进行审计。
@import url('https://fonts.googleapis.com/css2?family=Google+Sans:wght@400;500;600&family=Roboto:wght@300;400;500&display=swap');
/* Reset & base */
html, body, [class*="css"] {
font-family: 'Roboto', sans-serif;
margin: 0; padding: 0;
}
.stApp {
background: #0f0f0f;
color: #f1f1f1;
}
/* Hide streamlit chrome */
#MainMenu, footer, header { visibility: hidden; }
[data-testid="stSidebar"] { display: none; }
.block-container {
padding: 0 !important;
max-width: 100% !important;
}
/* ── TOP NAV BAR ── */
.topbar {
display: flex;
align-items: center;
justify-content: space-between;
background: #0f0f0f;
border-bottom: 1px solid #272727;
padding: 10px 20px;
position: sticky;
top: 0;
z-index: 100;
}
.topbar-left {
display: flex;
align-items: center;
gap: 12px;
}
.yt-logo {
font-size: 1.3rem;
font-weight: 700;
color: #fff;
letter-spacing: -0.5px;
}
.yt-logo span { color: #ff0000; }
.url-input-wrap {
flex: 1;
max-width: 600px;
margin: 0 24px;
}
/* ── MAIN TWO-PANEL LAYOUT ── */
.main-panels {
display: flex;
height: calc(100vh - 57px);
overflow: hidden;
}
/* Left: video panel */
.video-panel {
flex: 1;
background: #000;
display: flex;
flex-direction: column;
overflow: hidden;
}
.video-embed-wrap {
position: relative;
width: 100%;
padding-top: 56.25%; /* 16:9 */
background: #000;
flex-shrink: 0;
}
.video-embed-wrap iframe {
position: absolute;
top: 0; left: 0;
width: 100%; height: 100%;
border: none;
}
.video-info {
padding: 16px 20px;
border-top: 1px solid #272727;
background: #0f0f0f;
flex-shrink: 0;
}
.video-title {
font-family: 'Roboto', sans-serif;
font-size: 1.1rem;
font-weight: 500;
color: #f1f1f1;
margin: 0 0 4px 0;
line-height: 1.4;
}
.video-channel {
font-size: 0.82rem;
color: #aaa;
margin: 0;
}
/* Right: chat panel */
.chat-panel {
width: 400px;
min-width: 340px;
max-width: 420px;
background: #212121;
border-left: 1px solid #272727;
display: flex;
flex-direction: column;
overflow: hidden;
height: 100%;
}
.chat-header {
padding: 14px 18px 12px;
border-bottom: 1px solid #333;
flex-shrink: 0;
background: #212121;
}
.chat-header-top {
display: flex;
align-items: center;
justify-content: space-between;
margin-bottom: 2px;
}
.chat-title {
font-family: 'Roboto', sans-serif;
font-size: 1rem;
font-weight: 500;
color: #f1f1f1;
margin: 0;
}
.gemini-star {
font-size: 1.1rem;
margin-right: 6px;
}
.chat-subtitle {
font-size: 0.75rem;
color: #aaa;
margin-top: 2px;
}
/* Suggested questions */
.suggestions {
padding: 14px 16px 8px;
border-bottom: 1px solid #2d2d2d;
flex-shrink: 0;
}
.suggestions-label {
font-size: 0.78rem;
color: #aaa;
margin-bottom: 8px;
}
.suggestion-chips {
display: flex;
flex-direction: column;
gap: 6px;
}
.suggestion-chip {
background: transparent;
border: 1px solid #3d3d3d;
border-radius: 18px;
padding: 7px 14px;
font-size: 0.8rem;
color: #c8c8c8;
cursor: pointer;
text-align: right;
width: fit-content;
align-self: flex-end;
transition: background 0.15s, border-color 0.15s;
line-height: 1.3;
}
.suggestion-chip:hover {
background: #2d2d2d;
border-color: #555;
color: #f1f1f1;
}
/* Chat messages area */
.chat-messages {
flex: 1;
overflow-y: auto;
padding: 16px;
display: flex;
flex-direction: column;
gap: 14px;
scrollbar-width: thin;
scrollbar-color: #3d3d3d #212121;
}
/* Thinking dots */
@keyframes thinking-pulse {
0%, 80%, 100% { opacity: 0.2; transform: scale(0.8); }
40% { opacity: 1; transform: scale(1.1); }
}
.thinking-dot {
display: inline-block;
width: 7px; height: 7px;
border-radius: 50%;
background: #666;
animation: thinking-pulse 1.2s ease-in-out infinite;
}
/* Timestamp dropdown chip — pure CSS, no JS */
.ts-dropdown {
position: relative;
display: inline-block;
vertical-align: middle;
margin: 0 2px;
}
.ts-chip {
display: inline-flex;
align-items: center;
gap: 3px;
background: linear-gradient(180deg, #1c2a3a 0%, #142233 100%);
border: 1px solid #2a3f5a;
border-radius: 5px;
padding: 1px 8px;
font-size: 0.78rem;
font-family: 'Roboto Mono', monospace;
color: #8ab4f8;
cursor: pointer;
user-select: none;
white-space: nowrap;
transition: background 0.12s;
box-shadow: 0 1px 4px rgba(0,0,0,0.35);
}
.ts-chip:hover { background: #253549; color: #b0ccff; }
.ts-menu {
display: none;
position: absolute;
bottom: 100%;
margin-bottom: 6px;
left: 0;
background: #141a24;
border: 1px solid #2b3a52;
border-radius: 8px;
min-width: 210px;
z-index: 9999;
overflow: hidden;
box-shadow: 0 10px 24px rgba(0,0,0,0.55);
}
.ts-dropdown:hover .ts-menu,
.ts-dropdown:focus-within .ts-menu {
display: block;
}
.ts-menu::after {
content: "";
position: absolute;
bottom: -6px;
left: 0;
width: 100%;
height: 6px;
}
.ts-option {
display: block;
padding: 9px 14px;
font-size: 0.82rem;
color: #cfd8ea;
text-decoration: none;
cursor: pointer;
transition: background 0.12s;
white-space: nowrap;
}
.ts-option:hover { background: #223248; color: #fff; }
.ts-option + .ts-option { border-top: 1px solid #2d3a50; }
.chat-messages::-webkit-scrollbar { width: 4px; }
.chat-messages::-webkit-scrollbar-track { background: #212121; }
.chat-messages::-webkit-scrollbar-thumb { background: #3d3d3d; border-radius: 2px; }
/* Message bubbles */
.msg-user {
align-self: flex-end;
background: #2d2d2d;
border-radius: 18px 18px 4px 18px;
padding: 10px 14px;
max-width: 85%;
font-size: 0.87rem;
color: #f1f1f1;
line-height: 1.5;
word-wrap: break-word;
}
.msg-ai-wrap {
align-self: flex-start;
max-width: 95%;
display: flex;
flex-direction: column;
gap: 6px;
}
.msg-ai-label {
font-size: 0.72rem;
color: #888;
display: flex;
align-items: center;
gap: 4px;
margin-bottom: 2px;
}
.msg-ai {
background: transparent;
font-size: 0.87rem;
color: #e0e0e0;
line-height: 1.6;
word-wrap: break-word;
}
.msg-ai a {
color: #8ab4f8;
text-decoration: none;
}
.msg-ai a:hover { text-decoration: underline; }
/* Timestamp source chips */
.source-chips {
display: flex;
flex-wrap: wrap;
gap: 5px;
margin-top: 4px;
}
.inline-ts {
display: inline-flex;
align-items: center;
gap: 3px;
background: #1e2a3a;
border: 1px solid #2a3f5a;
border-radius: 5px;
padding: 1px 7px;
font-size: 0.78rem;
font-family: 'Roboto Mono', monospace;
color: #8ab4f8;
cursor: pointer;
transition: background 0.12s, transform 0.1s;
user-select: none;
white-space: nowrap;
vertical-align: middle;
margin: 0 2px;
}
.inline-ts:hover { background: #253549; color: #b0ccff; transform: translateY(-1px); }
.inline-ts:active { transform: translateY(0); background: #2a3f5a; }
/* Chat input area */
.chat-input-area {
padding: 10px 14px 8px;
border-top: 1px solid #2d2d2d;
background: #212121;
flex-shrink: 0;
}
.chat-disclaimer {
text-align: center;
font-size: 0.67rem;
color: #666;
padding: 4px 0 0;
}
/* Streamlit input overrides */
.stTextInput > div > div > input {
background: #2d2d2d !important;
border: 1px solid #3d3d3d !important;
border-radius: 22px !important;
color: #f1f1f1 !important;
font-size: 0.87rem !important;
padding: 10px 18px !important;
font-family: 'Roboto', sans-serif !important;
}
.stTextInput > div > div > input:focus {
border-color: #555 !important;
box-shadow: none !important;
outline: none !important;
}
.stTextInput > div > div > input::placeholder { color: #888 !important; }
.stButton > button {
background: transparent;
border: none;
color: #8ab4f8;
font-size: 0.85rem;
font-weight: 500;
padding: 6px 12px;
border-radius: 4px;
cursor: pointer;
font-family: 'Roboto', sans-serif;
transition: background 0.15s;
}
.stButton > button:hover { background: #2d2d2d; color: #c0d4ff; }
/* Empty / loading states */
.empty-state {
flex: 1;
display: flex;
flex-direction: column;
align-items: center;
justify-content: center;
text-align: center;
padding: 32px 24px;
color: #888;
}
.empty-icon { font-size: 2rem; margin-bottom: 10px; }
.empty-text { font-size: 0.85rem; line-height: 1.6; }
/* Top URL bar inputs */
div[data-testid="stHorizontalBlock"] .stTextInput > div > div > input {
background: #121212 !important;
border: 1px solid #303030 !important;
border-radius: 22px !important;
color: #f1f1f1 !important;
font-size: 0.88rem !important;
padding: 9px 16px !important;
}
/* Spinner */
.stSpinner > div { border-top-color: #aaa !important; }
/* Chips (suggestion buttons) styled via st.button with key trick */
div[data-suggestion="true"] .stButton > button {
background: transparent !important;
border: 1px solid #3d3d3d !important;
border-radius: 18px !important;
color: #c8c8c8 !important;
font-size: 0.8rem !important;
padding: 7px 14px !important;
width: 100% !important;
text-align: right !important;
justify-content: flex-end !important;
}
"""
app.py - YouTube AI Chat — Streamlit frontend
Two-panel layout: embedded video left, chat right
"""
import streamlit as st
import streamlit.components.v1 as components
from openai import OpenAI
import time
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
from transcript import extract_video_id, fetch_transcript, chunk_transcript, format_timestamp, make_youtube_link
from metadata import fetch_metadata
from embedder import build_index
from keyword_index import build_keyword_index
from chat import chat_with_video
# ── Page config ────────────────────────────────────────────────────────────────
st.set_page_config(
page_title="YT Chat",
page_icon="🎬",
layout="wide",
initial_sidebar_state="collapsed",
)
# ── Custom CSS ─────────────────────────────────────────────────────────────────
def load_local_css(filename: str):
css_path = os.path.join(os.path.dirname(__file__), filename)
if os.path.exists(css_path):
with open(css_path, "r", encoding="utf-8") as f:
st.markdown(f"<style>{f.read()}</style>", unsafe_allow_html=True)
else:
st.warning(f"Missing CSS file: {css_path}")
load_local_css("style.css")
# ── Query param handler is already set up above ──────────────────────────────────────
# ── Session state ──────────────────────────────────────────────────────────────
def init_state():
defaults = {
"videos": {},
"active_video_id": None,
"conversations": {},
"client": None,
"pending_input": "",
"awaiting_answer": "", # question waiting for LLM response
"processing_answer": False,
}
for k, v in defaults.items():
if k not in st.session_state:
st.session_state[k] = v
init_state()
# Handle jump-to-timestamp from HTML button click
if "jump_to" in st.query_params:
try:
jump_seconds = st.query_params.get("jump_to")
if jump_seconds:
st.session_state.jump_to_seconds = int(jump_seconds)
# Clear the param to avoid re-triggering
params = dict(st.query_params)
del params["jump_to"]
st.query_params.clear()
for k, v in params.items():
st.query_params[k] = v
except Exception as e:
print(f"Error handling jump_to param: {e}")
def get_client():
return st.session_state.client
def active_video():
vid = st.session_state.active_video_id
if vid and vid in st.session_state.videos:
return st.session_state.videos[vid]
return 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] = []
if vid:
return st.session_state.conversations[vid]
return []
# ── TOP NAV ────────────────────────────────────────────────────────────────────
st.markdown("""
<div class="topbar">
<div style="font-size:1.25rem;font-weight:700;color:#fff;letter-spacing:-0.3px;">
<span style="color:#ff0000;">▶</span> YT Chat
</div>
</div>
""", unsafe_allow_html=True)
# Controls row below topbar
ctrl_col1, ctrl_col2, ctrl_col3 = st.columns([4, 1, 1])
with ctrl_col1:
yt_url = st.text_input(
"url", placeholder="Paste YouTube URL...",
label_visibility="collapsed", key="url_input"
)
with ctrl_col2:
load_btn = st.button("Load Video", use_container_width=True)
with ctrl_col3:
# Show loaded videos selector if multiple
if len(st.session_state.videos) > 1:
video_options = {v["meta"]["title"][:28] + "…": k
for k, v in st.session_state.videos.items()}
selected_label = st.selectbox(
"Switch", list(video_options.keys()),
label_visibility="collapsed"
)
st.session_state.active_video_id = video_options[selected_label]
elif len(st.session_state.videos) == 1:
st.markdown(
f'<div style="font-size:0.75rem;color:#888;padding:8px 0;">1 video loaded</div>',
unsafe_allow_html=True
)
# Handle load
if load_btn:
url_val = yt_url.strip()
if not OPENAI_API_KEY:
st.error("OPENAI_API_KEY not found in .env file.")
st.stop()
elif not url_val:
st.error("Paste a YouTube URL.")
else:
if not st.session_state.client:
st.session_state.client = OpenAI(api_key=OPENAI_API_KEY)
video_id = extract_video_id(url_val)
if not video_id:
st.error("Couldn't parse a video ID from that URL.")
elif video_id in st.session_state.videos:
st.session_state.active_video_id = video_id
st.success("Already loaded — switched to it.")
st.rerun()
else:
prog = st.progress(0, text="Fetching metadata...")
try:
meta = fetch_metadata(video_id)
prog.progress(15, text="Fetching transcript...")
raw = fetch_transcript(video_id)
prog.progress(40, text="Chunking transcript...")
chunks = chunk_transcript(raw)
prog.progress(60, text=f"Embedding {len(chunks)} chunks...")
index, chunks = build_index(chunks, get_client())
prog.progress(80, text="Building keyword index...")
keyword_index = build_keyword_index(chunks)
prog.progress(95, text="Almost done...")
st.session_state.videos[video_id] = {
"meta": meta, "chunks": chunks,
"index": index, "keyword_index": keyword_index,
"chunk_count": len(chunks),
}
st.session_state.active_video_id = video_id
st.session_state.conversations[video_id] = []
prog.progress(100, text="Ready!")
time.sleep(0.3)
prog.empty()
st.rerun()
except ValueError as e:
prog.empty()
st.error(str(e))
except Exception as e:
prog.empty()
st.error(f"Error: {e}")
st.markdown("<div style='height:1px;background:#272727;margin:0;'></div>", unsafe_allow_html=True)
# ── MAIN TWO-PANEL LAYOUT ──────────────────────────────────────────────────────
video = active_video()
if not video:
# Empty state
st.markdown("""
<div style="display:flex;align-items:center;justify-content:center;
height:calc(100vh - 120px);flex-direction:column;
text-align:center;color:#555;gap:12px;">
<div style="font-size:3rem;">▶</div>
<div style="font-size:1rem;color:#888;font-weight:500;">Paste a YouTube URL above to get started</div>
<div style="font-size:0.82rem;color:#555;max-width:380px;line-height:1.6;">
Chat with any video — answers grounded in the transcript with clickable timestamps
</div>
</div>
""", unsafe_allow_html=True)
else:
meta = video["meta"]
chunks = video["chunks"]
index = video["index"]
video_id = meta["video_id"]
conversation = active_conversation()
# ── Two columns: video | chat ──────────────────────────────────────────────
left_col, right_col = st.columns([1.15, 0.85], gap="small")
# ── LEFT: Video embed + info ───────────────────────────────────────────────
with left_col:
origin = "http://localhost:8501"
# Create placeholder for video panel to allow re-rendering on jump
video_placeholder = st.empty()
# Check if we need to update start time
start_time = 0
if hasattr(st.session_state, 'jump_to_seconds') and st.session_state.jump_to_seconds:
start_time = st.session_state.jump_to_seconds
st.session_state.jump_to_seconds = None # Reset for next jump
with video_placeholder.container():
st.markdown(f"""
<div class="video-panel">
<div class="video-embed-wrap">
<iframe
id="yt-player"
name="yt-player"
src="https://www.youtube.com/embed/{video_id}?rel=0&modestbranding=1&enablejsapi=1&autoplay=1&start={start_time}&origin={origin}"
allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
allowfullscreen>
</iframe>
</div>
<div class="video-info">
<div class="video-title">{meta['title']}</div>
<div class="video-channel">{meta['author']}</div>
<div class="meta-row" style="display:flex;align-items:center;gap:12px;">
<span class="badge">{video['chunk_count']} chunks indexed</span>
<a class="yt-link" href="{meta['url']}" target="_blank">↗ Open on YouTube</a>
</div>
</div>
</div>
""", unsafe_allow_html=True)
# ── RIGHT: Chat panel ──────────────────────────────────────────────────────
with right_col:
# Clear input field if flag is set from previous submission
if st.session_state.get("should_clear_input", False):
st.session_state.chat_input = ""
st.session_state.should_clear_input = False
# Chat header
st.markdown("""
<div style="background:#212121;border:1px solid #2d2d2d;border-radius:10px;
overflow:hidden;display:flex;flex-direction:column;">
<div style="padding:14px 18px 10px;border-bottom:1px solid #2d2d2d;">
<div style="display:flex;align-items:center;gap:6px;">
<span style="font-size:1rem;">💬</span>
<span style="font-size:0.95rem;font-weight:500;color:#f1f1f1;">Ask about this video</span>
</div>
<div style="font-size:0.73rem;color:#777;margin-top:2px;">
Answers grounded in transcript · click timestamps to jump
</div>
</div>
</div>
""", unsafe_allow_html=True)
# Suggested questions + chat container
# Height shrinks when suggestions are visible so input stays on screen
suggestions = [ "Summarize this video", "What are the main topics discussed?", "What are the key takeaways?", ]
if st.session_state.get("processing_answer"):
st.markdown(
"<style>div[data-suggestion='true']{display:none !important;}</style>",
unsafe_allow_html=True
)
if not conversation:
is_thinking = bool(st.session_state.get("processing_answer"))
chat_container = st.container(height=230)
with chat_container:
if is_thinking:
question_text = st.session_state.get("awaiting_answer", "")
st.markdown(f"""
<div style="display:flex;justify-content:flex-end;margin:6px 0;">
<div style="background:#2d2d2d;border-radius:18px 18px 4px 18px;
padding:10px 14px;max-width:88%;font-size:0.86rem;
color:#f1f1f1;line-height:1.5;word-wrap:break-word;">
{question_text}
</div>
</div>
<div style="padding:8px 4px;">
<div style="font-size:0.7rem;color:#777;margin-bottom:6px;">✦ AI Assistant</div>
<div style="display:flex;align-items:center;gap:6px;">
<span class="thinking-dot"></span>
<span class="thinking-dot" style="animation-delay:.2s"></span>
<span class="thinking-dot" style="animation-delay:.4s"></span>
<span style="margin-left:4px;font-size:0.8rem;color:#555;">Thinking...</span>
</div>
</div>
""", unsafe_allow_html=True)
else:
st.markdown('<div data-suggestion="true">', unsafe_allow_html=True)
st.markdown("""
<div style="font-size:0.76rem;color:#888;margin-bottom:8px;padding:2px 2px 0;">
Not sure what to ask? Choose something:
</div>
""", unsafe_allow_html=True)
for i, s in enumerate(suggestions):
if st.button(s, key=f"suggestion_{i}", use_container_width=True):
st.session_state.pending_input = s
st.rerun()
st.markdown("""
<div style="text-align:center;color:#444;font-size:0.8rem;
padding:14px 16px 4px;line-height:1.7;">
Hello! Curious about what you're watching?<br>I'm here to help.
</div>
""", unsafe_allow_html=True)
st.markdown("</div>", unsafe_allow_html=True)
else:
# Conversation active — full height container, no suggestions
chat_container = st.container(height=420)
with chat_container:
for turn_idx, turn in enumerate(conversation):
if turn["role"] == "user":
st.markdown(f"""
<div style="display:flex;justify-content:flex-end;margin:6px 0;">
<div style="background:#2d2d2d;border-radius:18px 18px 4px 18px;
padding:10px 14px;max-width:88%;font-size:0.86rem;
color:#f1f1f1;line-height:1.5;word-wrap:break-word;">
{turn['content']}
</div>
</div>
""", unsafe_allow_html=True)
else:
import re as _re
reply_raw = turn["content"]
ts_pattern = r'\[([\d]{1,2}:[\d]{2}(?::[\d]{2})?)\]\((https://www\.youtube\.com/watch\?v=[\w-]+&t=(\d+)s?)\)'
def _timestamp_to_seconds(label: str) -> int:
parts = [int(p) for p in label.split(":")]
if len(parts) == 2:
return parts[0] * 60 + parts[1]
return parts[0] * 3600 + parts[1] * 60 + parts[2]
def _extract_timestamps(text: str):
extracted = []
# 1) Markdown links: [12:34](https://www.youtube.com/watch?v=...&t=754s)
for m in _re.finditer(ts_pattern, text):
extracted.append({"label": m.group(1), "seconds": m.group(3)})
cleaned = _re.sub(ts_pattern, '', text)
# 2) Plain timestamps: (12:34), 12:34, or 1:02:33
plain_ts_pattern = r'(?<!\d)(\d{1,2}:[0-5]\d(?::[0-5]\d)?)(?!\d)'
for m in _re.finditer(plain_ts_pattern, cleaned):
label = m.group(1)
seconds = str(_timestamp_to_seconds(label))
extracted.append({"label": label, "seconds": seconds})
# Remove parenthesized plain timestamps from text once captured.
cleaned = _re.sub(r'\(\s*\d{1,2}:[0-5]\d(?::[0-5]\d)?\s*\)', '', cleaned)
# Deduplicate while preserving order.
deduped = []
seen = set()
for item in extracted:
key = (item["label"], item["seconds"])
if key in seen:
continue
seen.add(key)
deduped.append(item)
return cleaned.strip(), deduped
def _format_block(text: str) -> str:
block = _re.sub(r'\*\*(.+?)\*\*', r'<strong>\1</strong>', text)
block = _re.sub(r'(?m)^\s*(\d+)[.)]\s+', lambda mm: f'<strong>{mm.group(1)}.</strong> ', block)
block = _re.sub(r'(?m)^\s*[-•]\s+', '• ', block)
return block.replace('\n', '<br>')
paragraphs = [p for p in reply_raw.split('\n\n') if p.strip()]
parsed_sections = []
for para in paragraphs:
numbered_starts = list(_re.finditer(r'(?m)^\s*\d+[.)]\s+', para))
# If paragraph contains multiple numbered items, split it into
# per-item sections so each item can get its own timestamp row.
if len(numbered_starts) >= 2:
intro_text = para[:numbered_starts[0].start()].strip()
if intro_text:
intro_display, intro_timestamps = _extract_timestamps(intro_text)
parsed_sections.append({
"text": intro_display,
"timestamps": intro_timestamps,
"is_timestamp_only": bool(intro_timestamps) and not intro_display,
})
for i, match in enumerate(numbered_starts):
start = match.start()
end = numbered_starts[i + 1].start() if i + 1 < len(numbered_starts) else len(para)
item_text = para[start:end].strip()
item_display, item_timestamps = _extract_timestamps(item_text)
parsed_sections.append({
"text": item_display,
"timestamps": item_timestamps,
"is_timestamp_only": bool(item_timestamps) and not item_display,
})
continue
para_display, para_timestamps = _extract_timestamps(para)
parsed_sections.append({
"text": para_display,
"timestamps": para_timestamps,
"is_timestamp_only": bool(para_timestamps) and not para_display,
})
# Redistribute timestamp-only paragraphs when possible so timestamps
# appear under each answer section (especially for numbered lists).
render_sections = []
sec_idx = 0
while sec_idx < len(parsed_sections):
section = parsed_sections[sec_idx]
if section["is_timestamp_only"]:
if render_sections:
render_sections[-1]["timestamps"].extend(section["timestamps"])
sec_idx += 1
continue
lines = [ln.strip() for ln in section["text"].splitlines() if ln.strip()]
numbered_lines = [ln for ln in lines if _re.match(r'^\d+[.)]\s+.+', ln)]
next_is_ts_only = (
sec_idx + 1 < len(parsed_sections)
and parsed_sections[sec_idx + 1]["is_timestamp_only"]
)
if numbered_lines and next_is_ts_only and not section["timestamps"]:
ts_pool = parsed_sections[sec_idx + 1]["timestamps"]
if len(ts_pool) >= len(numbered_lines):
for i, line in enumerate(numbered_lines):
render_sections.append({
"text": line,
"timestamps": [ts_pool[i]],
})
if len(ts_pool) > len(numbered_lines):
render_sections[-1]["timestamps"].extend(ts_pool[len(numbered_lines):])
sec_idx += 2
continue
render_sections.append({
"text": section["text"],
"timestamps": section["timestamps"][:],
})
sec_idx += 1
st.markdown("<div style='margin:6px 0;'><div style='font-size:0.7rem;color:#777;margin-bottom:4px;display:flex;align-items:center;gap:4px;'><span>✦</span> AI Assistant</div></div>", unsafe_allow_html=True)
for para_idx, section in enumerate(render_sections):
if not section["text"] and not section["timestamps"]:
continue
if section["text"]:
para_html = _format_block(section["text"])
st.markdown(
f'<div style="font-size:0.86rem;color:#e0e0e0;line-height:1.6;word-wrap:break-word;margin-bottom:6px;">{para_html}</div>',
unsafe_allow_html=True
)
if section["timestamps"]:
button_cols = st.columns(len(section["timestamps"]), gap='small')
for idx, ts_data in enumerate(section["timestamps"]):
label = ts_data["label"]
seconds = ts_data["seconds"]
with button_cols[idx]:
if st.button(
f'⏱ {label}',
key=f'ts_btn_{video_id}_{turn_idx}_{para_idx}_{idx}_{label}_{seconds}',
use_container_width=True
):
st.session_state.jump_to_seconds = int(seconds)
st.rerun()
# Show thinking bubble inside container if answer is pending
if st.session_state.get("processing_answer"):
st.markdown("""
<div style="margin:6px 0;">
<div style="font-size:0.7rem;color:#777;margin-bottom:4px;
display:flex;align-items:center;gap:4px;">
<span>✦</span> AI Assistant
</div>
<div style="display:flex;align-items:center;gap:6px;
color:#555;font-size:0.82rem;padding:4px 0;">
<span class="thinking-dot"></span>
<span class="thinking-dot" style="animation-delay:.2s"></span>
<span class="thinking-dot" style="animation-delay:.4s"></span>
<span style="margin-left:4px;">Thinking...</span>
</div>
</div>
""", unsafe_allow_html=True)
# Input row
inp_col, btn_col = st.columns([5, 1])
with inp_col:
user_input = st.text_input(
"Ask", placeholder="Ask a question...",
label_visibility="collapsed", key="chat_input"
)
with btn_col:
send_btn = st.button("→", key="send_btn")
# ── Two-phase send ────────────────────────────────────────────────────
# Phase 1: user submits → store question, append to convo, rerun immediately
# so the question bubble appears before LLM is called.
# Phase 2: awaiting_answer is set → call LLM, append reply, rerun.
# Detect new submission
new_question = ""
if st.session_state.pending_input:
new_question = st.session_state.pending_input
st.session_state.pending_input = ""
elif send_btn and user_input.strip():
new_question = user_input.strip()
if new_question:
# Phase 1 — show question immediately
conversation.append({"role": "user", "content": new_question, "sources": []})
st.session_state.awaiting_answer = new_question
st.session_state.processing_answer = True
st.session_state.should_clear_input = True # Flag to clear on next rerun
st.rerun()
# Phase 2 — question is visible, now generate the answer
if st.session_state.awaiting_answer and st.session_state.processing_answer:
question = st.session_state.awaiting_answer
history_for_api = [
{"role": t["role"], "content": t["content"]}
for t in conversation
if not (t["role"] == "user" and t["content"] == question and t == conversation[-1])
]
try:
keyword_index = video.get("keyword_index")
reply, sources = chat_with_video(
question, history_for_api,
index, chunks, video_id, get_client(),
keyword_index=keyword_index,
)
conversation.append({"role": "assistant", "content": reply, "sources": sources})
except Exception as e:
conversation.append({
"role": "assistant",
"content": f"Sorry, something went wrong: {e}",
"sources": []
})
finally:
st.session_state.awaiting_answer = ""
st.session_state.processing_answer = False
st.rerun()
# Disclaimer
st.markdown("""
<div style="text-align:center;font-size:0.67rem;color:#555;padding:4px 0 2px;">
AI can make mistakes, so double-check it.
</div>
""", unsafe_allow_html=True)
# Clear chat
if conversation:
if st.button("Clear chat", key="clear_chat"):
st.session_state.conversations[video_id] = []
st.rerun()
运行应用程序
在“运行应用”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
前提条件
在准备阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个具体职责，而非整个复杂的流程。对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不能保证业务的完整性。
安装依赖项
在“安装依赖项”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务上的完整性。
cd medium/chat-with-video
pip install -r requirements.txt
运行应用程序
在“运行应用”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能测试结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于会消耗资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。
streamlit run app.py
What are the main topics discussed?
What is mentioned about the relation of America with France?
后续改进方向
在下一阶段的改进中，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审核。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
结论
在结论阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品的一部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。
参考资料
在“参考阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
操作检查清单
在“操作检查清单阶段”，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境切换到共享环境时出现意外账单。