بث الكلام منخفض زمن الاستجابة من وكلاء الذكاء الاصطناعي باستخدام Kokoro TTS على Amazon SageMaker AI

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

عمليات دمج تحويل النص إلى كلام تخلق هذه الفجوة تحديداً. قد تُولَّد استجابة من جملة واحدة في 600 ميلي ثانية، لكن الاستجابة الكاملة تستغرق خمس ثوانٍ، وينتظر التكامل اكتمالها بالكامل قبل أن يبدأ التركيب.

يوضح هذا المنشور كيفية سد تلك الفجوة. نقوم ببث كل جملة إلى نموذج تحويل النص إلى كلام (TTS) Kokoro-82M مستضاف على Amazon SageMaker AI بمجرد أن ينتجها الوكيل، بحيث يبدأ التركيب من الجملة الأولى بدلاً من الاستجابة المكتملة. وإعادة هذا الصوت إلى المتصفح تتطلب حلاً خاصاً بها أيضاً، لذا نغطي كلا وسيلتي النقل للقيام بذلك وتكلفة كل منهما عملياً.

نظرة عامة على الحل

يغلق هذا الحل الفجوة عن طريق بث التركيب بالتوازي مع التوليد بدلاً من انتظار انتهائه. كلما أنتج الوكيل كل جملة، تُبَث تلك الجملة فوراً إلى نموذج تحويل النص إلى كلام Kokoro-82M مستضاف، بحيث يمكن أن يبدأ تشغيل الصوت قبل أن ينتهي الوكيل من إنتاج استجابته الكاملة. يحتوي Kokoro-82M على 82 مليون معامل فقط، وهو صغير بما يكفي للتشغيل بتكلفة منخفضة على مثيل GPU واحد متوسط الفئة. يحافظ ذلك على انخفاض كل من زمن الاستجابة وتكلفة الاستدلال لوكيل صوتي يتعامل مع حركة مرور حقيقية.

يعمل Kokoro-82M في حاوية مخصصة على نقطة نهاية SageMaker AI في الوقت الفعلي، على مثيل ml.g5.xlarge GPU. تعرض الحاوية مسارين: POST /invocations، وهو يبث استجابة لمدخل واحد كامل، و WS /invocations-bidirectional-stream، وهو يبقي اتصالاً واحداً مفتوحاً عبر رسائل متعددة في الاتجاهين.

يوضح المخطط التالي مسار الطلب من العميل إلى النموذج:

الشكل 1: مسار الطلب من العميل إلى نقطة النهاية في الوقت الفعلي لـ SageMaker AI عبر تدفق ثنائي الاتجاه

المتطلبات الأساسية

قبل أن تبدأ، تحتاج إلى:

• حساب AWS مع أذونات لإنشاء نقاط نهاية SageMaker AI؛ وAgentCore runtime، وهي إمكانية من Amazon Bedrock AgentCore؛ وأدوار AWS Identity and Access Management (IAM)؛ ومستودع Amazon Elastic Container Registry (Amazon ECR)، بما في ذلك كحد أدنى إجراءات sagemaker:CreateEndpoint, sagemaker:CreateEndpointConfig, sagemaker:CreateModel, sagemaker:InvokeEndpointWithBidirectionalStreamو bedrock:InvokeModelWithResponseStream .
• تثبّت الحاوية النموذجية إصداراً أقدم من PyTorch مع إشعار أمني معروف (PYSEC-2026-2286، تم إصلاحه في torch 2.10.0). راجع وحدّث التثبيت قبل أي استخدام يتجاوز إثبات القيمة.
• AWS Cloud Development Kit (AWS CDK) v2، مثبَّت ومُهيَّأ (bootstrapped) في حسابك ومنطقتك AWS المستهدفتين.
• نقطة نهاية SageMaker AI في الوقت الفعلي منشورة من حاوية تُعلن عن تسمية Docker com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true .
• Python 3.12 أو أحدث، و uv، وDocker لبناء صورة الحاوية (عميل HTTP/2 الأساسي لا يوفر سوى ملفات wheels مبنية مسبقاً لـ Python 3.12 وأحدث).
• AWS SDK for Python (Boto3) مع دعم لـ invoke_endpoint_with_bidirectional_stream.
• الإلمام بـ WebSockets وStarlette أو إطار ASGI مماثل.
• كود النشر من مستودع GitHub المشار إليه في هذه التدوينة.

الشرح التفصيلي

يبدأ الشرح التفصيلي بنشر نقطة النهاية والوكيل، ثم يتناول التغييرات من جهة العميل التي تجعل البث التزايدي يعمل من البداية إلى النهاية.

نشر نقطة النهاية والوكيل

يتم نشر نقطة نهاية Kokoro-82M وبيئة تشغيل الوكيل عبر أدايتين من أدوات سطر الأوامر CLI، تُغلّف كل منهما حزم AWS CDK الخاصة بها:

uv run speech-infra deploy kokoro-82m
uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>

speech-infra deploy تُوفّر نقطة نهاية Kokoro-82M. يشمل ذلك حزم الأساس ومخزن النماذج المؤقت التي تنشئ أدوار IAM الأساسية وتزامن أوزان النموذج إلى Amazon Simple Storage Service (Amazon S3). ثم تقوم حزمة نقطة النهاية ببناء صورة الحاوية وإنشاء نموذج SageMaker AI وإعدادات نقطة النهاية ونقطة النهاية نفسها. agent-infra deploy تُوفّر بيئة تشغيل Amazon Bedrock AgentCore التي تستضيف وكيل Strands، والذي يستدعي نقطة النهاية التي نشرتها للتو، لذا انشر نقطة النهاية أولاً.

الخطوة 1: التعامل مع نوع إطار WebSocket الذي يرسله SageMaker AI فعلياً

تنقل طبقة النقل ثنائية الاتجاه التي يستخدمها SageMaker AI كل جزء من رسالة العميل إلى حاويتك كإطار WebSocket ثنائي (binary). من الطرق الشائعة لكتابة معالج WebSocket في Starlette هو استدعاء receive_text()، التي تقرأ message["text"] وتُطلق استثناءً KeyError إذا لم يكن ذلك المفتاح موجوداً. وبما أن كل رسالة يرسلها هذا العميل تصل بصيغة ثنائية، فإن هذا المعالج يفشل في كل طلب.

يتم تشخيص هذا الفشل خطأً بشكل شائع. معالج استثناءات عام يعيد التوجيه str(e) إلى العميل يحول KeyError('text') إلى إطار خطأ يقرأ 'text'، وهو ما يبدو كحقل مفقود، لا كمشكلة في تأطير الرسائل. والنقطة (endpoint) نفسها تُبلّغ عن InService طوال الوقت، لأنها تقبل الاتصالات وتستجيب لكل رسالة، مع خطأ في كل مرة.

اقبل كلا نوعي الإطار صراحةً بدلاً من افتراض أن أحدهما سيظهر:

from starlette.websockets import WebSocket, WebSocketDisconnect

async def _receive_message(websocket: WebSocket) -> str:
    message = await websocket.receive()
    if message["type"] == "websocket.disconnect":
        raise WebSocketDisconnect(message.get("code", 1000))
    payload = message.get("text")
    if payload is None:
        payload = message.get("bytes", b"").decode("utf-8")
    return payload

يقوم هذا بفك ترميز البايتات كـ UTF-8 عندما لا يكون هناك حقل نصي موجود، بدلاً من افتراض نوع إطار واحد. اختبر ذلك بإرسال إطار ثنائي (binary frame)، لأن هذا النمط من الفشل ينتج إطار خطأ مُحكم البناء وحالة نقطة (endpoint) سليمة، لا انهياراً واضحاً.

الخطوة 2: إبقاء الجلسة مفتوحة بين الرسائل

بعد أن ينتهي الحاوية (container) من توليد صوت رسالة واحدة، يكون الغريزي هو إغلاق الجلسة، تماماً كما تنتهي دورة HTTP. هذا خطأ هنا. بعد إرسال صوت الرسالة وإطار تحكم synthesis_complete ، ينبغي أن تعود حلقة الحاوية إلى الانتظار على receive() للرسالة التالية على نفس الجلسة التي لا تزال مفتوحة. أغلق الاتصال فقط عندما يرسل العميل رسالة {"type": "close"} صريحة.

يوضح المخطط التالي جلسة واحدة تحمل جملتين على التوالي، مع الإغلاق في النهاية فقط:

الشكل 2: جلسة ثنائية الاتجاه واحدة تحمل جملتين على التوالي، مع الإغلاق في النهاية فقط

بهذه الطريقة، يمكنك إرسال عدة جمل، واحدة تلو الأخرى، على جلسة دائمة واحدة، دون أي كلفة لإعادة الاتصال بينها. الإغلاق المتسرع بعد رد واحد سيفرض اتصالاً جديداً لكل جملة، مما يُبطل معظم فائدة زمن الاستجابة المنخفض.

الخطوة 3: بث الجمل أثناء إنتاجها من قبل الوكيل

مع وضع سلوك الجلسة في مكانه، فإن الإصلاح من جهة العميل لمشكلة زمن الاستجابة هو فئة (class) توحّد النص الواصل في شكلين: سلسلة كاملة معروفة مسبقاً، أو تدفق رموز (token stream) حي. وهي تنتج المخرجات نفسها في الحالتين: جمل كاملة، جاهزة للإرسال بمجرد اكتمال كل منها.

import re

_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")


class IncrementalSentenceChunker:
    def __init__(self) -> None:
        self._buffer = ""

    def feed(self, fragment: str) -> list[str]:
        self._buffer += fragment
        parts = _SENTENCE_RE.split(self._buffer)
        if len(parts) == 1:
            return []
        self._buffer = parts[-1]
        return [s.strip() for s in parts[:-1] if s.strip()]

    def flush(self) -> str | None:
        remainder = self._buffer.strip()
        self._buffer = ""
        return remainder or None

استدعِ feed() مرة واحدة لكل مقطع نصي عند وصوله. وتعيد هذه الدالة أي جمل مكتملة حديثاً، مع حجز النص اللاحق. واستدعِ flush() مرة واحدة، في نهاية الإدخال، لإطلاق أي نص جزئي متبقٍ حتى بدون علامة ترقيم لاحقة. هذا يمنع تدفقاً ينتهي قبل وصول نقطة من إسقاط جملة أخيرة.

هذا ليس توليداً صوتياً متزامناً. كل معالج ثنائي الاتجاه هو حلقة تسلسلية صارمة: استلام رسالة، توليد صوتها، إرسال ذلك الصوت، إرسال synthesis_complete، ثم استلام الرسالة التالية. صوت الجملة الثانية لا يُولَّد بينما لا تزال الجملة الأولى تُرسل. ما يحققه الإرسال التدريجي أضيق نطاقاً: يبدأ توليد صوت الجملة الأولى لحظة وجودها، بدلاً من انتظار الرد الكامل. بالنسبة لرد متعدد الجملات مع وقت توليد حقيقي بين جملته الأولى والأخيرة، فهذه هي مكسب زمن الاستجابة بأكمله.

تحققنا من هذا مقابل نقطة نهاية حية بإرسال إدخال من ثلاث جمل عبر جلسة واحدة. وصلت المقاطع الثلاثة كلها مع صوت يطابق النص المرسل لكل منها:

الخطوة 4: اختيار وسيلة نقل لإعادة الصوت إلى العميل

بث النص إلى النموذج تدريجياً يحل نصف المشكلة. النصف الآخر هو إيصال الصوت الناتج إلى أيٍّ سيعرضه، وعادةً ما يكون ذلك متصفحاً. معظم واجهات الوكلاء والدردشة تُبث النص بالفعل عبر Server-Sent Events (SSE): اتجاه واحد، ومدعومة أصلياً في معظم المتصفحات. بروتوكول SSE هو بروتوكول نصي، لم يُبنَ لحمل حمولة ثنائية خام. إن فريقاً يضيف الصوت إلى خط SSE قائم يلجأ بطبيعته إلى النمط الذي يملكه بالفعل: ترميز الصوت بـ base64 ولفّه في حدث JSON.

قمنا بقياس تكلفة ذلك الأسلوب عبر 24 مقطعًا، وتأكدنا عبر Amazon Transcribe من أن الجودة لم تختلف بين الترميزات، وقارنا MP3 مع AAC:

يؤدي Base64 إلى تضخيم كل حمولة بنسبة تقارب 33% المتوقعة. التكلفة الأكثر خفيةً هي محاذاة الإطارات. تُحدَّد أحداث SSE بسطر فارغ، لكن الصوت بداخلها يُقسَّم إلى إطارات MP3. ولا تتطابق المخططان بالضرورة مع الطريقة التي يقسِّم بها النقل البيانات إلى أجزاء. في استجابة تم قياسها، وصلت أربعة إطارات MP3 منطقية في ست أجزاء تسليم، اثنان منها ينتهيان في منتصف إطار. المستهلك الذي يتعامل مع كل جزء تسليم كوحدة كاملة واحدة يفشل مع حركة بيانات عادية سليمة. الحل هو تخزين البايتات الواردة مؤقتًا والتصرف فقط عند العثور على محدد:

buffer = b""
for event in resp["Body"]:
    if "PayloadPart" not in event:
        continue
    buffer += event["PayloadPart"]["Bytes"]
    while b"\n\n" in buffer:
        frame, buffer = buffer.split(b"\n\n", 1)
        handle(frame)

البديل المستخدم اليوم للاتصال المواجه للنموذج يتخطى base64 وتأطير SSE تمامًا. فهو يضاعف الصوت الثنائي الخام وإطارات تحكم JSON عبر نفس الجلسة ثنائية الاتجاه التي تحمل بالفعل إدخال النص التدريجي، ويميّز بينهما بقاعدة واحدة.

import json


def _decode_control_frame(chunk: bytes) -> dict | None:
    try:
        decoded = json.loads(chunk.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return decoded if isinstance(decoded, dict) else None

الجزء الذي يفشل في فك ترميزه ككائن JSON هو صوت، يُمرَّر كما ورد، دون أي خطوة ترميز في أي مكان على المسار. هناك حالة حدّية واحدة مهمة: الصوت الخام PCM يمكن أن يبدأ مشروعًا بالبايت 0x7b، وهو الحرف ASCII {، وهو نفس الحرف الذي يفتح كائن JSON. هذه القاعدة تتجنب التصنيف الخاطئ لذلك الصوت لأنها تتطلب تحليل الجزء بأكمله كـ JSON صالح، وليس بايته الأول وحده، وهو ما لن يفعله إطار PCM عمليًا أبدًا.

ملاحظة: إذا كان واجهتك الأمامية لديها بالفعل مسار SSE يعمل للنص، فإن الصوت المشفر بـ base64 داخل حدث SSE هو الخيار العملي. تكلفته حوالي 33 بالمئة حمولة إضافية وخطوة تخزين مؤقت وإعادة تجميع، لكنه يعيد استخدام بنية تحتية تشغّلها بالفعل. أما إذا كنت تبني مسار الصوت من الصفر، أو كانت زمن الاستجابة أولوية، فإن نهج الثنائي وJSON الموحد يتجنب كلتا التكلفتين ويعيد استخدام الاتصال الذي يحمل النص بالفعل.

الخطوة 5: تجميع كل شيء في وكيل حقيقي

الحالة الأشمل من البداية إلى النهاية هي وكيل توليدي مباشر يصيغ استجابته رمزًا تلو الآخر، دون أي وسيلة لمعرفة الجملة الرابعة قبل وجود الجملة الأولى. يعمل هذا المثال كوكيل Strags مدعوم بـ Amazon Bedrock، ومنشور على Amazon Bedrock AgentCore runtime، ويستدعي نقطة نهاية Kokoro بالنمط من الخطوة 4. ثلاثة مسارات اتصال تتصل في هذا التصميم: المسار المواجه للمتصفح الذي يحمل Server-Sent Events إلى المتصل، واستدعاء البث من الوكيل إلى Amazon Bedrock لتوليد النص، والجلسة ثنائية الاتجاه للوكيل مع نقطة نهاية Kokoro للتركيب الصوتي.

يوضح المخطط التالي مسارات الاتصال الثلاثة تلك، ومكان تداخل التوليد والتركيب الصوتي:

الشكل 3: مسارات الاتصال الثلاثة التي تربط المتصل وAmazon Bedrock ونقطة نهاية Kokoro

تحمل حمولة الطلب حقل mode ، إما "bidi" أو "batch"، ويُوجَّه إلى مساري كود داخل نفس دالة نقطة الدخول. المقارنة المعيارية بينهما تغيّر متغيرًا واحدًا تمامًا: متى يبدأ التركيب الصوتي. في run_bidi، تعمل مهمتان بشكل متزامن. الأولى تستنزف تدفق نص Bedrock وتغذي الجمل المكتملة إلى المُقسِّم. والثانية تستنزف جلسة Kokoro وتُصدر أجزاء الصوت فور وصولها. كلتاهما تدفعان إلى قائمة انتظار مشتركة واحدة، بحيث يُصدر المستهلك الأحداث بترتيب الوصول الحقيقي: يمكن أن يصل صوت جملة مبكرة إلى المتصل بينما لا يزال Bedrock يولّد جملة لاحقة. run_batch هو العكس المتعمد: ينتظر الاستجابة الكاملة قبل استدعاء تركيب صوتي واحد، ويُبقي كأساس عادل للمقارنة.

وفي الاتجاه المعاكس، المسار من حاوية الوكيل عائدة إلى المتصل يستخدم بالفعل base64، لسبب مختلف عمّا تصفه الخطوة 4. BedrockAgentCoreAppتتطلب نقطة دخول أن تُصدر قواميس JSON قابلة للتسلسل بشكل عادي، يحوّلها SDK الخاص بها تلقائيًا إلى data: {json}\n\n إطارات. لا توجد طريقة لتسليمها إطارًا ثنائيًا خامًا. هذه نتيجة لبيئة التشغيل المُدارة الأصلية بـ JSON، وليست الخيار الذي قاسته الخطوة 4 وابتعدت عنه. كلا run_bidi و run_batch يدفعان تلك التكلفة المتطابقة على هذا المسار، لذا فإن الفرق المقاس بينهما هو توقيت بحت: متى يُصدر أول جزء صوت.

الاختبار والتحقق

عزلّا مساهمة نقل TTS نفسها في زمن الوصول إلى أول صوت عبر إرسال نفس النص المكون من 267 حرفًا وأربع جمل بطريقتين: كرسالة واحدة (دفعي)، وكجملته الأولى فقط (تدريجي). أجرينا خمس محاولات لكل منهما مقابل نقطة النهاية المباشرة.

كما قيس سابقًا، أدى إرسال الجملة الأولى فقط إلى خفض متوسط زمن الوصول إلى أول بايت صوتي من 329.3 ms إلى 156.7 ms، أي تحسّن نسبي بنسبة 52.4 بالمئة، معزول عن نقل TTS نفسه.

ثم قسنا الحالة الأشمل من البداية إلى النهاية بنموذج لغة حقيقي ينتج النص، بتشغيل نفس المطالبة عبر run_bidi و run_batch مقابل نفس نموذج Bedrock ونقطة النهاية المباشرة، بحيث يكون المتغير الوحيد هو متى يبدأ التركيب الصوتي.

باستخدام نفس المنهجية، قيس زمن الوصول إلى أول صوت في وضع bidi بسرعة 2.4 مرة أسرع من الوضع الدفعي مقابل حاوية محلية، وبسرعة 3.5 مرة أسرع مقابل بيئة تشغيل AgentCore المنشورة.

يوضح المخطط التالي تلك الفجوة في كلتا البيئتين:

الشكل 4: الزمن النسبي للوصول إلى أول صوت لوضعي الدفعي وbidi، محليًا وعلى بيئة التشغيل المنشورة

تتسع الفجوة، بدلاً من أن تضيق، في حالة النشر المُوزَّع بمجرد إضافة قفزة شبكة حقيقية. run_bidi يستوعب تلك القفزة داخل التداخل الذي يستغلّه أصلًا بين التوليد والتخليق. run_batch لا يملك تداخلاً مكافئًا يخفي تلك التكلفة فيه، لأنه ينفّذ شيئًا واحدًا في كل مرة، لذا فإن كل ملّي ثانية إضافية من حمل الشبكة تقع مباشرة على انتظاره التسلسلي.

لإعادة إنتاج هذه المقارنة مقابل نشر خود، شغّل نفس اختبار bidi مقابل batch الذي تعرضه حاوية الوكيل:

uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"

يرسل هذا المطالبة عبر كلا الوضعين ويعرض الطابع الزمني لأول audio_chunk حدث مقابل agent_start لكل منهما. إذا ظهرت لك أخطاء بدلاً من ذلك، فتحقق من KeyError: 'text' عرَض الخطوة 1. نقطة نهاية تُبلغ عن InService بينما يفشل كل طلب تشير إلى هذا الخلل، وليس إلى مشكلة في النشر.

التنظيف

لتجنب الرسوم المستمرة، قم بحذف الموارد التي ينشئها هذا الدليل التطبيقي بترتيب عكسي لعملية النشر: الوكيل (agent) أولاً، ثم نقطة النهاية (endpoint):

uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id>
uv run speech-infra destroy kokoro-82m

agent-infra destroy يُزيل مورد بيئة تشغيل Amazon Bedrock AgentCore. speech-infra destroy يُزيل نموذج SageMaker AI وإعداد نقطة النهاية ونقطة النهاية. ولا يقوم أي منهما بإفراغ حاوية Amazon S3 التي تخزّن أوزان النموذج مؤقتاً أو بحذف مستودع Amazon ECR الذي يحتوي على صورة الحاوية، إذ يمكن مشاركة كليهما عبر عدة نماذج. إذا لم تعد بحاجة إليهما، قم بإفراغ حاوية Amazon S3 وحذفها وإزالة مستودع Amazon ECR يدوياً، لأن نسخة النموذج المحفوظة وصورة الحاوية تُبقي على تحميل رسوم تخزين صغيرة حتى تتم إزالتهما.

الخاتمة

أظهرنا كيفية إزالة أحد مصادر زمن الاستجابة الذي يمكن تجنبه من وكيل صوتي، وهو انتظار اكتمال استجابة الوكيل بالكامل قبل بدء توليف الكلام، وكيفية قياس تكلفة ثانية أكثر خفوة: الصوت المُرمَّز بـ base64 داخل SSE يضخّم حجم الحمولة بنحو الثلث، وقد يؤدي إلى اختلال المحاذاة مع حدود إطارات MP3 إذا لم تأخذ التخزين المؤقت اللازم في الحسبان. وقد أدى بث كل جملة إلى نقطة نهاية ثنائية الاتجاه في SageMaker AI فور إنتاج الوكيل لها إلى تقليص الوقت حتى أول صوت إلى أقل من النصف في العزل، وبمقدار 2.4 إلى 3.5 مرة في وكيل منشور فعلياً. وإن إرسال الإطارات الثنائية للصوت وإطارات تحكم JSON عبر تلك الجلسة نفسها، بدلاً من ترميز الصوت كنص، يُلغي تضخّم الحجم والعمل الإضافي المطلوب لإعادة محاذاة الإطارات الذي يتطلبه مرحّل SSE تقليدي بسيط.

إذا كنت تضيف الصوت إلى وكيل يبث النص بالفعل عبر SSE، فابدأ بأسلوب base64 في الخطوة 4، ثم أعد النظر في النقل الموحّد عندما يصبح زمن الاستجابة أولوية. لمعرفة المزيد، راجع وثائق الاستدلال الفوري في Amazon SageMaker AI و وثائق Amazon Bedrock AgentCore. وإذا كنت ترغب في الحصول على مساعدة في تطبيق هذا النمط على بنيتك التحتية الخاصة، فتواصل مع فريق حساب AWS الخاص بك.

عن المؤلف