RAG ও LLM পাইপলাইনের জন্য ভিডিও ট্রান্সক্রিপ্ট

সংক্ষিপ্ত উত্তর: ট্রান্সক্রিপ্ট সাধারণ টেক্সট হিসেবে নয়, JSON হিসেবে আনুন। segments অ্যারেতে প্রতিটি ক্যাপশন লাইনের শুরু ও শেষের সময় থাকে, তাই আপনি সময় অনুযায়ী অংশে ভাগ করতে পারেন এবং শুরুর সময়টি মেটাডেটা হিসেবে সংরক্ষণ করতে পারেন। তখন আপনার মডেলের প্রতিটি উত্তর ভিডিওর নির্দিষ্ট মুহূর্তের লিংকে যেতে পারে।

সময়যুক্ত সেগমেন্ট কেন সাধারণ টেক্সটের চেয়ে ভালো#

• উদ্ধৃতি: যে অংশ জানে সেটি 754 সেকেন্ডে শুরু হয়, সেটি ওই মুহূর্তের লিংকে পরিণত হয়।
• স্বাভাবিক সীমানা: ক্যাপশন লাইন বিরতিতে শেষ হয়, তাই অংশ খুব কমই কোনো বাক্যকে মাঝখানে কাটে।
• উৎসের তথ্য: source_origin বলে দেয় টেক্সটটি নির্মাতার, প্ল্যাটফর্মের স্বয়ংক্রিয় ক্যাপশনের, নাকি AI ট্রান্সক্রিপশনের, আর language বলে দেয় এর ভাষা।

আনুন ও অংশে ভাগ করুন#

এটি একটি ট্রান্সক্রিপ্ট আনে এবং সেগমেন্টগুলো প্রায় এক মিনিটের উইন্ডোতে গুচ্ছবদ্ধ করে। ট্রান্সক্রিপ্ট JSON এর গঠন লেখা আছে API রেফারেন্সে।

import requests
 
API = "https://www.transcriptdock.com"
H = {"Authorization": "Bearer td_live_YOUR_KEY"}
 
def transcript(url: str) -> dict:
    job = requests.post(f"{API}/v1/jobs", headers={**H, "Idempotency-Key": f"rag-{url}"[:128]},
        json={"source": {"url": url}, "mode": "captions_only"}).json()
    while job["status"] not in ("succeeded", "failed", "cancelled"):
        job = requests.get(f"{API}/v1/jobs/{job['id']}", params={"wait": 25}, headers=H).json()
    if job["status"] != "succeeded":
        raise RuntimeError(job["error"]["code"])
    return requests.get(f"{API}/v1/transcripts/{job['result_id']}", headers=H).json()
 
def chunks(t: dict, seconds: float = 60.0):
    """Group caption segments into windows of about `seconds`, never splitting a segment."""
    url = t["source"]["canonical_url"]
    buf, start = [], None
    for seg in t["segments"]:
        if start is None:
            start = seg["start"]
        buf.append(seg["text"])
        if seg["end"] - start >= seconds:
            yield {"text": " ".join(buf), "start": start, "url": f"{url}&t={int(start)}s" if "youtube" in url else url}
            buf, start = [], None
    if buf:
        yield {"text": " ".join(buf), "start": start, "url": url}
 
t = transcript("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
for c in chunks(t):
    print(c["start"], c["url"], c["text"][:80])  # embed c["text"], store start and url as metadata

অনেক ভিডিও#

• ভিডিও খোঁজা দিয়ে ভিডিও খুঁজুন: YouTube খোঁজা, একটি চ্যানেল, একটি প্লেলিস্ট বা একটি TikTok প্রোফাইল।
• এগুলো একটি ব্যাচ হিসেবে সাবমিট করুন। একটি ওয়েবহুক প্রতিটি কাজ শেষ হলে আপনার ইনজেস্ট প্রক্রিয়াকে জানিয়ে দেয়, যাতে বারবার যাচাই করতে না হয়।
• ট্রান্সক্রিপ্ট আবার পড়া ও আবার এক্সপোর্ট করা বিনামূল্যে, এবং যে লিংক আগে ট্রান্সক্রাইব করেছেন তা আবার সাবমিট করলে সংরক্ষিত ফলাফল ফেরত আসে, তাই ইনজেস্ট আবার চালালে দুবার খরচ হয় না।

চাহিদামতো আনা এজেন্ট#

প্রতিটি পাইপলাইনের ইনডেক্স দরকার হয় না। Transcript Dock MCP সার্ভার এর সাথে যুক্ত একটি এজেন্ট, কোনো ব্যবহারকারী একটি ভিডিওর কথা বললে get_video_transcript কল করতে পারে এবং টেক্সট সরাসরি তার কনটেক্সটে পড়ে নিতে পারে। ফ্রেমওয়ার্কের জন্য দেখুন LangChain ও n8n।