بناء وكيل محادثة ذكي قابل للتطوير باستخدام Pipecat و Hugging Face

يقدم هذا البرنامج التعليمي شرحًا عمليًا لبناء وكيل محادثة ذكي (AI) كامل الوظائف من الصفر باستخدام إطار عمل Pipecat. سنتناول بالتفصيل كيفية إعداد خط أنابيب (Pipeline) يربط بين فئات معالجة الإطارات (FrameProcessor) المخصصة، حيث تكون إحداها مسؤولة عن معالجة مدخلات المستخدم وتوليد الردود باستخدام نموذج Hugging Face، والأخرى مسؤولة عن تنسيق وعرض تدفق المحادثة. سنقوم أيضًا بتنفيذ مُولِّد مدخلات المحادثة (ConversationInputGenerator) لمحاكاة الحوار، واستخدام PipelineRunner و PipelineTask لتنفيذ تدفق البيانات بشكل غير متزامن. يوضح هذا الهيكل كيفية تعامل Pipecat مع المعالجة القائمة على الإطارات، مما يُمكِّن من التكامل النمطي للمكونات مثل نماذج اللغات، ومنطق العرض، والإضافات المستقبلية مثل وحدات الكلام.

1. التثبيت والبدء

نبدأ بتثبيت المكتبات المطلوبة، بما في ذلك Pipecat و Transformers و PyTorch، ثم نُنشئ عمليات الاستيراد الخاصة بنا. نقوم باستيراد مكونات Pipecat الأساسية، مثل Pipeline و PipelineRunner و FrameProcessor، بالإضافة إلى واجهة برمجة التطبيقات (API) pipeline من Hugging Face لتوليد النص. هذا يُحضّر بيئتنا لبناء وتشغيل وكيل المحادثة الذكي بسلاسة.

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

2. معالج المحادثة البسيط (SimpleChatProcessor)

نقوم بتنفيذ SimpleChatProcessor، والذي يقوم بتحميل نموذج Hugging Face DialoGPT-small لتوليد النص ويحافظ على سجل المحادثة للسياق. مع وصول كل إطار نصي (TextFrame)، نقوم بمعالجة مدخلات المستخدم، وتوليد استجابة النموذج، وتنظيفها، ودفعها للأمام في خط أنابيب Pipecat للعرض. يضمن هذا التصميم أن وكيل الذكاء الاصطناعي لدينا يمكنه إجراء محادثات متماسكة ومتعددة الأدوار في الوقت الفعلي.

class SimpleChatProcessor(FrameProcessor):
    """معالج محادثة ذكي بسيط باستخدام 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!"
                    else:
                        ai_response = "I'm not sure I understand. Can you rephrase?"
                    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)

3. معالج عرض النص (TextDisplayProcessor) ومولد مدخلات المحادثة (ConversationInputGenerator)

نقوم بإنشاء TextDisplayProcessor لتنسيق وعرض استجابات الذكاء الاصطناعي بشكل مرتب، مع تتبع عدد مرات التبادل في المحادثة. بالإضافة إلى ذلك، يقوم ConversationInputGenerator بمحاكاة سلسلة من رسائل المستخدم كأجسام TextFrame، مع إضافة فترات توقف قصيرة بينها لمحاكاة تدفق ذهابًا وإيابًا طبيعيًا أثناء العرض التجريبي.

class TextDisplayProcessor(FrameProcessor):
    """يعرض إطارات النص بتنسيق محادثة"""
    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:
    """يولد مدخلات محادثة تجريبية"""
    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 < len(self.demo_conversations) - 1:
                await asyncio.sleep(2)

4. وكيل الذكاء الاصطناعي البسيط (SimpleAIAgent) والوظيفة الرئيسية (main)

في SimpleAIAgent، نقوم بربط كل شيء معًا من خلال دمج معالج المحادثة، ومعالج العرض، ومولد المدخلات في خط أنابيب Pipecat واحد. تقوم طريقة run_demo بتشغيل PipelineRunner لمعالجة الإطارات بشكل غير متزامن بينما يقوم مولد المدخلات بتغذية رسائل المستخدم المُحاكاة. يسمح هذا الإعداد المُنسق للوكيل بمعالجة المدخلات، وتوليد الردود، وعرضها في الوقت الفعلي، لإكمال تدفق المحادثة الشامل.

class SimpleAIAgent:
    """وكيل محادثة ذكي بسيط باستخدام 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()

هذا البرنامج التعليمي يوضح كيفية بناء وكيل محادثة ذكي قابل للتطوير باستخدام Pipecat و Hugging Face. يمكنك بناء عليه لإضافة المزيد من الميزات المتقدمة.

المصدر: MarkTechPost