Pipecat इंटीग्रेशन

यह गाइड बताती है कि Speech Engine ब्रेन सर्वर में LLM पाइपलाइन के रूप में Pipecat का इस्तेमाल कैसे करें। Speech Engine वॉइस लूप — स्पीच-टू-टेक्स्ट, टर्न-टेकिंग और टेक्स्ट टू स्पीच — संभालता है, जबकि Pipecat प्रोसेसर्स की एक कॉम्पोज़ेबल पाइपलाइन (LLM कॉल, RAG, फंक्शन कॉल, गार्डरेल्स, कंटेंट फ़िल्टर) के ज़रिए टेक्स्ट जनरेशन संभालता है।

आर्किटेक्चर

Speech Engine SDK बाहरी लेयर के रूप में चलता है — जब भी यूज़र बोलना खत्म करता है, उसका on_transcript कॉलबैक चलता है। कॉलबैक के अंदर, आप एक Pipecat पाइपलाइन बनाते हैं, बातचीत का इतिहास LLMContextFrame के रूप में देते हैं और पाइपलाइन का टेक्स्ट आउटपुट वापस Speech Engine पर स्ट्रीम करते हैं। ElevenLabs टेक्स्ट को स्पीच में बदलता है और यूज़र को सुनाता है।

Pipecat पाइपलाइन सिर्फ़ एक टर्न की अवधि तक चलती है। नया ट्रांसक्रिप्ट आने पर, अगली पाइपलाइन शुरू होने से पहले पिछली पाइपलाइन रद्द हो जाती है — इसी तरह Speech Engine का इंटरप्शन हैंडलिंग पाइपलाइन तक पहुंचता है।

इस पैटर्न का इस्तेमाल कब करें

जब आपके ब्रेन को सिर्फ़ एक LLM कॉल से ज़्यादा की ज़रूरत हो, तब Pipecat उपयोगी है:

• रिट्रीवल-ऑगमेंटेड जनरेशन, फंक्शन कॉल या गार्डरेल्स के लिए कॉम्पोज़ेबल प्रोसेसर्स
• फ्रेम-आधारित मिडलवेयर, जो हर चरण में ट्रैफ़िक की जांच, बदलाव या उसे ब्लॉक कर सकता है
• कई एजेंट्स में साझा किए गए दोबारा इस्तेमाल होने वाले पाइपलाइन फ़्रैगमेंट्स

अगर आपका ब्रेन “ट्रांसक्रिप्ट इन, LLM कॉल आउट” है, तो Speech Engine क्विकस्टार्ट ज़्यादा आसान है। जब पाइपलाइन खुद अहम हिस्सा हो, तब Pipecat चुनें।

ज़रूरी शर्तें

• एक Speech Engine। इसे बनाने के लिए Speech Engine क्विकस्टार्ट फ़ॉलो करें।
• Python 3.10+ (pipecat-ai के लिए ज़रूरी)।
• ब्रेन सर्वर के लिए पब्लिक HTTPS टनल (जैसे ngrok)।

डिपेंडेंसीज़ इंस्टॉल करें

pip install "pipecat-ai[openai]" "elevenlabs" "python-dotenv"

pipecat-ai[openai] OpenAI LLM सर्विस को शामिल करता है। अगर चाहें, तो किसी दूसरे प्रोवाइडर के लिए एक्स्ट्रा बदलें (anthropic, google आदि)।

Pipecat ब्रेन बनाएं

ब्रेन के दो हिस्से हैं: एक TextSink प्रोसेसर, जो स्ट्रीम किए गए टेक्स्ट को asyncio.Queue में डालता है, और एक run_pipecat_brain कोरूटीन, जो एक-टर्न पाइपलाइन बनाता है और चंक्स को async iterator के रूप में देता है।

import asyncio
import os
from typing import AsyncIterator
from dotenv import load_dotenv
from pipecat.frames.frames import (
    Frame,
    LLMContextFrame,
    LLMFullResponseEndFrame,
    LLMTextFrame,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.openai.llm import OpenAILLMService
load_dotenv()
SYSTEM_PROMPT = (
    "You are a helpful voice assistant. Keep responses concise and conversational."
)
class TextSink(FrameProcessor):
    """Drain LLMTextFrame text into an asyncio.Queue."""
    def __init__(self, queue: asyncio.Queue):
        super().__init__()
        self._queue = queue
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, LLMTextFrame):
            await self._queue.put(frame.text)
        elif isinstance(frame, LLMFullResponseEndFrame):
            await self._queue.put(None)  # sentinel
        await self.push_frame(frame, direction)
def build_messages(transcript: list[dict]) -> list[dict]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for turn in transcript:
        role = "assistant" if turn["role"] == "agent" else turn["role"]
        messages.append({"role": role, "content": turn["content"]})
    return messages
async def run_pipecat_brain(transcript: list[dict]) -> AsyncIterator[str]:
    """Yield response text chunks from a one-turn Pipecat pipeline."""
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o-mini",
    )
    queue: asyncio.Queue[str | None] = asyncio.Queue()
    sink = TextSink(queue)
    task = PipelineTask(Pipeline([llm, sink]))
    runner = PipelineRunner(handle_sigint=False)
    async def drive():
        context = LLMContext(build_messages(transcript))
        await task.queue_frame(LLMContextFrame(context))
        await task.stop_when_done()
    run_task = asyncio.create_task(runner.run(task))
    drive_task = asyncio.create_task(drive())
    try:
        while True:
            chunk = await queue.get()
            if chunk is None:
                break
            yield chunk
    finally:
        await task.cancel()
        await asyncio.gather(run_task, drive_task, return_exceptions=True)

पाइपलाइन में सिर्फ़ LLM सर्विस और सिंक होते हैं — कोई STT या TTS प्रोसेसर नहीं, क्योंकि इन्हें Speech Engine संभालता है। LLMContextFrame इनपुट है; LLMTextFrame चंक्स आउटपुट हैं।

run_pipecat_brain एक async जनरेटर है। हर दिया गया चंक सीधे Speech Engine पर जाता है, इसलिए पूरा रिस्पॉन्स तैयार होने से पहले ही एजेंट बोलना शुरू कर देता है।

इसे Speech Engine सर्वर से जोड़ें

Speech Engine SDK का send_response एक स्ट्रिंग या स्ट्रिंग्स का कोई भी async iterable स्वीकार करता है, इसलिए आप सीधे run_pipecat_brain(transcript) पास कर सकते हैं। Speech Engine से मिलने वाले ConversationMessage ऑब्जेक्ट्स को ब्रेन में भेजने से पहले सादे dicts में बदलें।

import asyncio
import os
from dotenv import load_dotenv
from elevenlabs import AsyncElevenLabs
from brain import run_pipecat_brain
load_dotenv()
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"]
async def on_transcript(transcript, session):
    history = [{"role": m.role, "content": m.content} for m in transcript]
    await session.send_response(run_pipecat_brain(history))
async def main():
    engine = await elevenlabs.speech_engine.get(SPEECH_ENGINE_ID)
    await engine.serve(
        port=3001,
        path="/ws",
        debug=True,
        on_transcript=on_transcript,
    )
if __name__ == "__main__":
    asyncio.run(main())

नया ट्रांसक्रिप्ट आने पर Speech Engine SDK पिछले टर्न का टास्क रद्द कर देता है, जिससे run_pipecat_brain के try/finally ब्लॉक के ज़रिए async जनरेटर और अंदर का PipelineTask रद्द हो जाता है।

सर्वर चलाएं

ngrok http 3001
python server.py

क्विकस्टार्ट में दिखाए गए उसी टोकन एंडपॉइंट और क्लाइंट कोड का इस्तेमाल करके ब्राउज़र से Speech Engine से कनेक्ट करें। Pipecat पाइपलाइन सर्वर साइड पर चलती है; ब्राउज़र को सामान्य Speech Engine बातचीत दिखाई देती है।

पाइपलाइन बढ़ाएं

टेक्स्ट-ओनली Pipecat पाइपलाइन में कोई भी फ्रेम प्रोसेसर शामिल हो सकता है, जो LLMTextFrame या LLMContextFrame पर काम करता हो। कुछ आम विकल्प:

• गार्डरेल्स: LLM से पहले रखा गया FrameProcessor, जो LLMContextFrame की जांच करता है और असुरक्षित कॉन्टेक्स्ट को बदलता या ब्लॉक करता है।
• फंक्शन कॉल्स: OpenAILLMService पर टूल्स रजिस्टर करें और Pipecat टूल-कॉल फ्रेम्स को नेटिव रूप से संभालता है। अंतिम असिस्टेंट टेक्स्ट फिर भी LLMTextFrame के रूप में आता है।
• मल्टी-स्टेज रीजनिंग: दो OpenAILLMService इंस्टेंस को चेन करें, जिनके बीच एक कस्टम प्रोसेसर हो जो दूसरे पास के लिए कॉन्टेक्स्ट फिर से लिखे।
• आउटपुट फ़िल्टरिंग: LLM के बाद रखा गया FrameProcessor, जो हर LLMTextFrame की जांच करता है और TextSink तक पहुंचने से पहले अस्वीकार्य कंटेंट को हटाता या फिर से लिखता है।

पाइपलाइन का आकार वही रहता है — Pipeline([processor_a, llm, processor_b, sink]) — और run_pipecat_brain में कोई बदलाव नहीं होता।

प्रोडक्शन के लिए ध्यान देने योग्य बातें

• कैंसलेशन सुरक्षा: अगर पाइपलाइन पूरी तरह शुरू होने से पहले PipelineTask.cancel() कॉल किया जाए, तो डेडलॉक हो सकता है (pipecat-ai/pipecat#4276)। ऊपर दिया गया try/finally पैटर्न सुरक्षित है, क्योंकि cancel() कम-से-कम एक फ्रेम क्यू होने के बाद ही चलता है।
• प्रॉम्प्ट इंजेक्शन: स्पीच-टू-टेक्स्ट आउटपुट यूज़र इनपुट है। इसे LLM को देने से पहले ट्रांसक्रिप्ट को वैलिडेट या नॉर्मलाइज़ करें, खासकर जब कोई डाउनस्ट्रीम प्रोसेसर टेक्स्ट का इस्तेमाल टूल कॉल या डेटाबेस क्वेरी में करता हो।
• ब्रेन सर्वर ऑथेंटिकेशन: आपके /ws एंडपॉइंट पर अनधिकृत कनेक्शन रोकने के लिए Speech Engine पर साझा सीक्रेट सेट करें और ब्रेन सर्वर में उसकी जांच करें:

  await elevenlabs.speech_engine.update(
      speech_engine_id=SPEECH_ENGINE_ID,
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
• LLM प्रोवाइडर: pipecat-ai[openai] में OpenAILLMService शामिल है। Anthropic के लिए pipecat-ai[anthropic] इंस्टॉल करें और AnthropicLLMService का इस्तेमाल करें; बाकी पाइपलाइन में कोई बदलाव नहीं होगा।

अगले चरण