# 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"],
<a21>"title": r["title"],</a22>"channel": r["channel"],</a23>"url": r["url"],</a24>})</a25>corpus_docs.extend(docs)</a26>return corpus_docs</a27></a28>def research(query: str, k=12):</a29>corpus = build_corpus(query)</a30>if not corpus:</a31>return "No transcripts available."</a32>vs = VectorStore</a33>vs.add(corpus)</a34>passages = vs.search(query, k=k)</a35>md = call_claude(query, passages)</a36>timestamp = datetime.utcnow.isoformat</a37>return f"<!-- generated {timestamp} UTC -->\n\n" + md</a38></a39>if __name__ == "__main__":</a40>print(research("LLM agents for YouTube research"))</a40>