Implementation Guide for Building Modular Dialogue AI Agents with Pipelines and Hugs

!pip install -q pipecat-ai transformers torch accelerate numpy


import asyncio
import logging
from typing import AsyncGenerator
import numpy as np


print("🔍 Checking available Pipecat frames...")


try:
   from pipecat.frames.frames import (
       Frame,
       TextFrame,
   )
   print("✅ Basic frames imported successfully")
except ImportError as e:
   print(f"⚠️  Import error: {e}")
   from pipecat.frames.frames import Frame, TextFrame


from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


from transformers import pipeline as hf_pipeline
import torch

class SimpleChatProcessor(FrameProcessor):
   """Simple conversational AI processor using HuggingFace"""
   def __init__(self):
       super().__init__()
       print("🔄 Loading HuggingFace text generation model...")
       self.chatbot = hf_pipeline(
           "text-generation",
           model="microsoft/DialoGPT-small",
           pad_token_id=50256,
           do_sample=True,
           temperature=0.8,
           max_length=100
       )
       self.conversation_history = ""
       print("✅ Chat model loaded successfully!")


   async def process_frame(self, frame: Frame, direction: FrameDirection):
       await super().process_frame(frame, direction)
       if isinstance(frame, TextFrame):
           user_text = getattr(frame, "text", "").strip()
           if user_text and not user_text.startswith("AI:"):
               print(f"👤 USER: {user_text}")
               try:
                   if self.conversation_history:
                       input_text = f"{self.conversation_history} User: {user_text} Bot:"
                   else:
                       input_text = f"User: {user_text} Bot:"


                   response = self.chatbot(
                       input_text,
                       max_new_tokens=50,
                       num_return_sequences=1,
                       temperature=0.7,
                       do_sample=True,
                       pad_token_id=self.chatbot.tokenizer.eos_token_id
                   )


                   generated_text = response[0]["generated_text"]
                   if "Bot:" in generated_text:
                       ai_response = generated_text.split("Bot:")[-1].strip()
                       ai_response = ai_response.split("User:")[0].strip()
                       if not ai_response:
                           ai_response = "That's interesting! Tell me more."
                   else:
                       ai_response = "I'd love to hear more about that!"


                   self.conversation_history = f"{input_text} {ai_response}"
                   await self.push_frame(TextFrame(text=f"AI: {ai_response}"), direction)
               except Exception as e:
                   print(f"⚠️  Chat error: {e}")
                   await self.push_frame(
                       TextFrame(text="AI: I'm having trouble processing that. Could you try rephrasing?"),
                       direction
                   )
       else:
           await self.push_frame(frame, direction)

class TextDisplayProcessor(FrameProcessor):
   """Displays text frames in a conversational format"""
   def __init__(self):
       super().__init__()
       self.conversation_count = 0


   async def process_frame(self, frame: Frame, direction: FrameDirection):
       await super().process_frame(frame, direction)
       if isinstance(frame, TextFrame):
           text = getattr(frame, "text", "")
           if text.startswith("AI:"):
               print(f"🤖 {text}")
               self.conversation_count += 1
               print(f"    💭 Exchange {self.conversation_count} completen")
       await self.push_frame(frame, direction)




class ConversationInputGenerator:
   """Generates demo conversation inputs"""
   def __init__(self):
       self.demo_conversations = [
           "Hello! How are you doing today?",
           "What's your favorite thing to talk about?",
           "Can you tell me something interesting about AI?",
           "What makes conversation enjoyable for you?",
           "Thanks for the great chat!"
       ]


   async def generate_conversation(self) -> AsyncGenerator[TextFrame, None]:
       print("🎭 Starting conversation simulation...n")
       for i, user_input in enumerate(self.demo_conversations):
           yield TextFrame(text=user_input)
           if i 

class SimpleAIAgent:
   """Simple conversational AI agent using Pipecat"""
   def __init__(self):
       self.chat_processor = SimpleChatProcessor()
       self.display_processor = TextDisplayProcessor()
       self.input_generator = ConversationInputGenerator()


   def create_pipeline(self) -> Pipeline:
       return Pipeline([self.chat_processor, self.display_processor])


   async def run_demo(self):
       print("🚀 Simple Pipecat AI Agent Demo")
       print("🎯 Conversational AI with HuggingFace")
       print("=" * 50)


       pipeline = self.create_pipeline()
       runner = PipelineRunner()
       task = PipelineTask(pipeline)


       async def produce_frames():
           async for frame in self.input_generator.generate_conversation():
               await task.queue_frame(frame)
           await task.stop_when_done()


       try:
           print("🎬 Running conversation demo...n")
           await asyncio.gather(
               runner.run(task),     
               produce_frames(),    
           )
       except Exception as e:
           print(f"❌ Demo error: {e}")
           logging.error(f"Pipeline error: {e}")


       print("✅ Demo completed successfully!")

async def main():
   logging.basicConfig(level=logging.INFO)
   print("🎯 Pipecat AI Agent Tutorial")
   print("📱 Google Colab Compatible")
   print("🤖 Free HuggingFace Models")
   print("🔧 Simple & Working Implementation")
   print("=" * 60)
   try:
       agent = SimpleAIAgent()
       await agent.run_demo()
       print("n🎉 Tutorial Complete!")
       print("n📚 What You Just Saw:")
       print("✓ Pipecat pipeline architecture in action")
       print("✓ Custom FrameProcessor implementations")
       print("✓ HuggingFace conversational AI integration")
       print("✓ Real-time text processing pipeline")
       print("✓ Modular, extensible design")
       print("n🚀 Next Steps:")
       print("• Add real speech-to-text input")
       print("• Integrate text-to-speech output")
       print("• Connect to better language models")
       print("• Add memory and context management")
       print("• Deploy as a web service")
   except Exception as e:
       print(f"❌ Tutorial failed: {e}")
       import traceback
       traceback.print_exc()




try:
   import google.colab
   print("🌐 Google Colab detected - Ready to run!")
   ENV = "colab"
except ImportError:
   print("💻 Local environment detected")
   ENV = "local"


print("n" + "="*60)
print("🎬 READY TO RUN!")
print("Execute this cell to start the AI conversation demo")
print("="*60)


print("n🚀 Starting the AI Agent Demo...")


await main()