# file: yt_search.py
from googleapiclient.discovery import build
import os
YOUTUBE_API_KEY = os.environ — channel, date\n\n"
"---\n"
"JSON schema: {\"claims\":[{\"claim\":str,\"support\":[{\"video_id\":str,\"start\":float,\"end\":float}]}]}\n"
)
def call_claude(goal: str, passages: list[dict]):
passages_str = "\n\n".join(
f"[rank {p['rank']} | score {p['score']:.3f}] (vID={p.get('video_id','?')}, {p.get('start',0):.1f}-{p.get('end',0):.1f})\n{p['text']}"
for p in passages
)
msg = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1800,
temperature=0.2,
system=SYSTEM_PROMPT,
messages=[
{"role": "user", "content": USER_TEMPLATE.format(goal=goal, passages=passages_str)}
])
return msg.content[0].text

# file: run_agent.py
from yt_search import search_youtube
from transcripts import fetch_transcript
from chunking import transcript_to_docs
from embeddings import VectorStore
from orchestrator import call_claude
from datetime import datetime
def build_corpus(query: str, max_videos=8):
results = search_youtube(query, max_results=max_videos)
corpus_docs = []
for r in results:
tx = fetch_transcript(r["video_id"]) or []
if not tx:
continue
docs = transcript_to_docs(tx)
for d in docs:
d.update({
"video_id": r["video_id"]
"title": r["title"]
"channel": r["channel"]
"url": r["url"]
})
corpus_docs.extend(docs)
return corpus_docs
def research(query: str, k=12):
corpus = build_corpus(query)
if not corpus:
return "No transcripts available."
vs = VectorStore
vs.add(corpus)
passages = vs.search(query, k=k)
md = call_claude(query, passages)
timestamp = datetime.utcnow.isoformat
return f"<!-- generated {timestamp} UTC -->\n\n" + md
if __name__ == "__main__":
print(research("LLM agents for YouTube research"))