ملاحظات الاجتماعات باستخدام تحويل الكلام إلى نص

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

python notes.py standup.wav

على طول الطريق سنقوم بما يلي:

تفريغ تسجيل صوتي باستخدام /audio/transcriptions
طلب التوقيتات الزمنية، وهو ما لا يوفّره كل نموذج
استخراج القرارات والمهام الإجرائية وفق مخطط
التعامل مع حقيقة أن التفريغ لا يذكر أبدًا من يتحدّث
تقسيم تسجيل طويل دون فقدان الساعة الزمنية

الإعداد

تحتاج إلى Python 3.9 أو أحدث، وحزمة requests، ومفتاح Venice API. راجع إنشاء مفتاح API إن لم يكن لديك واحد. أحضر أي تسجيل لمحادثة بصيغة wav أو mp3 أو m4a أو flac أو aac أو mp4 أو ogg أو webm.

pip install requests
export VENICE_API_KEY="your-api-key-here"

from __future__ import annotations

import json
import os
import sys
import wave

import requests

BASE_URL = "https://api.venice.ai/api/v1"
AUTH = {"Authorization": f"Bearer {os.environ['VENICE_API_KEY']}"}
JSON_HEADERS = {**AUTH, "Content-Type": "application/json"}

1. تفريغ التسجيل

نقطة النهاية /audio/transcriptions متوافقة مع OpenAI وتستقبل رفعًا متعدّد الأجزاء (multipart). يجب أن يكون الملف جزءًا حقيقيًا من نوع ملف، إذ لا يُقبل ترميز base64 على هذه النقطة.

def transcribe(path: str, model: str, timestamps: bool = False) -> dict:
    with open(path, "rb") as audio:
        response = requests.post(
            f"{BASE_URL}/audio/transcriptions",
            headers=AUTH,
            files={"file": (os.path.basename(path), audio, "audio/wav")},
            data={
                "model": model,
                "response_format": "json",
                "timestamps": str(timestamps).lower(),
            },
            timeout=600,
        )
    response.raise_for_status()
    return response.json()

curl https://api.venice.ai/api/v1/audio/transcriptions \
  -H "Authorization: Bearer $VENICE_API_KEY" \
  -F "file=@./standup.wav" \
  -F "model=openai/whisper-large-v3" \
  -F "response_format=json" \
  -F "timestamps=true"

تُحتسب تكلفة التفريغ بحسب طول الصوت، لا بحسب مقدار ما قيل فيه، مما يجعل تكلفة الاجتماع سهلة التوقّع قبل تشغيله:

النموذج	السعر لكل ثانية صوت	ساعة كاملة من الاجتماع
`stt-xai-v1`	$0.0000315	$0.11
`nvidia/parakeet-tdt-0.6b-v3`	$0.0001	$0.36
`openai/whisper-large-v3`	$0.0001	$0.36
`elevenlabs/scribe-v2`	$0.000167	$0.60

استدعِ GET /models?type=asr للحصول على القائمة الحالية بدلًا من تثبيت هذه، إذ إن المكتبة تتغيّر.

2. اطلب التوقيتات

التوقيتات هي ما يجعل الملاحظات قابلة للتحقّق، ولذلك فإن هذا الخيار هو الأهم، والإعداد الافتراضي لن يعطيك إياها:

print(transcribe("standup.wav", "nvidia/parakeet-tdt-0.6b-v3", timestamps=True).keys())
print(transcribe("standup.wav", "openai/whisper-large-v3", timestamps=True).keys())

dict_keys(['text'])
dict_keys(['duration', 'text', 'timestamps'])

nvidia/parakeet-tdt-0.6b-v3 هو النموذج الافتراضي، وهو يقبل timestamps=true ثم يتجاهله. لا خطأ ولا تحذير، مجرد استجابة لا تحوي سوى text. إن احتجت إلى التوقيتات، اطلب نموذجًا يُعيدها وتحقّق من وجود المفتاح.

عندما يُعيد نموذج ما التوقيتات فعلًا، يكون timestamps كائنًا لا قائمة، ويعتمد المفتاح داخله على النموذج. يجمع Whisper حسب العبارة، بينما يجمع Scribe حسب الكلمة:

whisper = transcribe("standup.wav", "openai/whisper-large-v3", timestamps=True)
scribe = transcribe("standup.wav", "elevenlabs/scribe-v2", timestamps=True)

print(list(whisper["timestamps"]), json.dumps(whisper["timestamps"]["segment"][0]))
print(list(scribe["timestamps"]), json.dumps(scribe["timestamps"]["word"][0]))

["segment"] {"text": " Okay, let's keep this to 10 minutes. Where are we on the checkout migration?", "start": 0.21, "end": 4.21}
["word"] {"word": "Okay,", "start": 0.34, "end": 0.759}

المقاطع على مستوى العبارة هي الحجم الصحيح لهذه المهمة. توقيتات الكلمات مفيدة للترجمات المصاحبة، لكنها أدقّ من أن يُبنى عليها قرار. سنُسطّح هذه المقاطع إلى أسطر يسبقها الوقت، وهو كل ما يحتاجه النموذج للاستشهاد بها لاحقًا:

def timed_lines(transcription: dict, offset: float = 0.0) -> list[str]:
    segments = transcription.get("timestamps", {}).get("segment")
    if not segments:
        raise RuntimeError(
            "This model returned no segment timings. Use openai/whisper-large-v3."
        )
    return [
        f"[{segment['start'] + offset:.1f}s] {segment['text'].strip()}"
        for segment in segments
    ]

for line in timed_lines(whisper)[:4]:
    print(line)

[0.2s] Okay, let's keep this to 10 minutes. Where are we on the checkout migration?
[5.2s] Backend is done.
[6.5s] I finished the payment adapter yesterday and it's on staging.
[10.5s] The one thing I'm not sure about is whether we keep the old endpoint alive after cutover.

3. استخراج الملاحظات

صِف الملاحظات التي تريدها بوصفها مخططًا (schema)، فتكون النتيجة سجلًّا بدلًا من نصٍّ نثري عليك تحليله:

NOTES_SCHEMA = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "decisions": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "decision": {"type": "string"},
                    "spoken_at": {"type": "number", "description": "Seconds into the recording."},
                },
                "required": ["decision", "spoken_at"],
                "additionalProperties": False,
            },
        },
        "action_items": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "Name as spoken, or 'unassigned'."},
                    "task": {"type": "string"},
                    "due": {"type": "string", "description": "As stated, or 'not stated'."},
                    "spoken_at": {"type": "number"},
                },
                "required": ["owner", "task", "due", "spoken_at"],
                "additionalProperties": False,
            },
        },
        "open_questions": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["summary", "decisions", "action_items", "open_questions"],
    "additionalProperties": False,
}

SYSTEM = (
    "You turn meeting transcripts into notes. The transcript has no speaker labels, "
    "so attribute a task only when a name is spoken. Use 'unassigned' otherwise. "
    "spoken_at is the start time of the line the item came from."
)


def write_notes(lines: list[str], attendees: list[str] | None = None) -> dict:
    system = SYSTEM
    if attendees:
        system += (
            f" The attendees are {', '.join(attendees)}. Speech recognition often "
            "mangles names, so map what you hear to the closest attendee."
        )

    response = requests.post(
        f"{BASE_URL}/chat/completions",
        headers=JSON_HEADERS,
        json={
            "model": "zai-org-glm-5-2",
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": "\n".join(lines)},
            ],
            "response_format": {
                "type": "json_schema",
                "json_schema": {"name": "notes", "strict": True, "schema": NOTES_SCHEMA},
            },
            "temperature": 0,
            "max_completion_tokens": 2000,
            "venice_parameters": {
                "include_venice_system_prompt": False,
                "disable_thinking": True,
            },
        },
        timeout=300,
    )
    response.raise_for_status()
    choice = response.json()["choices"][0]
    if choice["finish_reason"] == "length":
        raise RuntimeError("Ran out of tokens. The JSON is truncated. Raise the budget.")
    return json.loads(choice["message"]["content"])

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

{
  "decisions": [
    { "decision": "Keep the old endpoint alive for two weeks after cutover, then remove it.", "spoken_at": 16.8 },
    { "decision": "Turn on the new checkout form for 10% of traffic on Monday; if error rate stays under 0.5%, increase to 50%.", "spoken_at": 34.5 }
  ],
  "action_items": [
    { "owner": "Tomas", "task": "Put the deprecation notice in the changelog by Friday.", "due": "Friday", "spoken_at": 19.6 },
    { "owner": "May", "task": "Own the frontend rollout of the new checkout form.", "due": "Monday", "spoken_at": 39.9 },
    { "owner": "unassigned", "task": "Ask legal to review the new refund copy and report back.", "due": "tomorrow", "spoken_at": 48.1 }
  ]
}

كل قيمة spoken_at حقيقية. اقفز إلى الثانية 19.6 لتسمع الجملة التي أنشأت المهمة.

إن أعطيت النموذج أسطرًا بلا توقيتات، فستعود كل قيمة spoken_at صفرًا. الحقل مطلوب، والنموذج لا يملك ما يضعه فيه، والحقل المطلوب توجيهٌ بإنتاج شيء، لا دعوة للإعلان عن الجهل. هذا يستحقّ التذكّر كلما بدا لك أن المخطط يعمل كما ينبغي: صحّة الشكل ليست صحّة القيم.

4. لا أحد مُصنَّف

في هذا الإخراج شيئان خاطئان، وكلاهما ينبع من المصدر ذاته. مسؤول إطلاق الميزة هو May. اسمها الحقيقي Mei. التعرّف على الكلام أقل موثوقيته في أسماء العلم، والأسماء هي بالضبط ما تحتاجه الإسناد، فهذا الفشل متوقّع لا مصادفة سيئة الحظ. البند الأخير unassigned، رغم أن شخصًا تعهّد به بوضوح. كان السطر: “I’ll ask legal today and report back tomorrow”، وقد سجّل التفريغ الكلمات دون تسجيل من قالها. هذه الثانية ليست علّة يمكنك إصلاحها. لا يُجري أي نموذج تفريغ في Venice تمييزًا للمتحدّثين (diarization)، فليس هناك حقل speaker تلجأ إليه في أيٍّ منها. التفريغ تيّار نصّ واحد بلا صوت، ولا يمكن إسناد المهام إلا حين يُنطق اسم صاحبها بصوت مسموع، كما في “Tomas, can you put the deprecation notice in the changelog”. أما الأولى فيمكنك إصلاحها بأن تُخبر النموذج بمن كان في الغرفة:

notes = write_notes(lines, attendees=["Priya Raman", "Tomas Vidal", "Mei Lin"])

Tomas Vidal   due=Friday    @ 19.6s  Put the deprecation notice in the changelog by Friday.
Mei Lin       due=Monday    @ 39.9s  Own the frontend rollout of the new checkout form.
unassigned    due=Tomorrow  @ 48.1s  Ask legal to review the new refund copy and report back.

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

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

5. أطول من طلب واحد

يبلغ الحدّ الأقصى للرفع 25 ميغابايت، وهو ما تصل إليه أسرع مما تتوقّع مع الصوت غير المضغوط، كما أن الاجتماع الطويل يستحقّ التقسيم على أي حال حتى لا يُكلّفك فشل واحد التفريغ بأكمله. بالنسبة لملفات WAV، تكفي المكتبة القياسية، ولا حاجة إلى ffmpeg:

def split_wav(path: str, chunk_seconds: int = 600) -> list[tuple[str, float]]:
    """Split into chunks, returning each path with its offset into the original."""
    chunks: list[tuple[str, float]] = []
    with wave.open(path, "rb") as source:
        rate = source.getframerate()
        stem = path.rsplit(".", 1)[0]
        index = 0
        while True:
            frames = source.readframes(rate * chunk_seconds)
            if not frames:
                break
            part = f"{stem}.part{index}.wav"
            with wave.open(part, "wb") as out:
                out.setnchannels(source.getnchannels())
                out.setsampwidth(source.getsampwidth())
                out.setframerate(rate)
                out.writeframes(frames)
            chunks.append((part, index * chunk_seconds))
            index += 1
    return chunks

الإزاحة (offset) هي بيت القصيد. يُفرَّغ كل جزء كما لو أنه يبدأ من الصفر، ولذلك يجب إعادة توقيتاته إلى الجدول الزمني للتسجيل الأصلي قبل أن يراها النموذج. هذا ما تستخدمه معامِلة offset في timed_lines:

def transcribe_long(path: str, model: str, chunk_seconds: int = 600) -> list[str]:
    lines: list[str] = []
    for part, offset in split_wav(path, chunk_seconds):
        lines.extend(timed_lines(transcribe(part, model, timestamps=True), offset))
        os.remove(part)
    return lines

قسّم اجتماع “standup” نفسه إلى قطع طولها عشرون ثانية، فتبقى الساعة صادقةً عبر نقاط الوصل. أما الكلمات فلا:

[16.8s] Let's keep it for two weeks, then remove it.
[19.6s] Thomas?
[20.0s] awesome.
[20.4s] Can you put the deprecation notice in the change log by Friday?
[24.1s] Yes, I'll do that.

جملة واحدة قيلت هناك: “Tomas, can you put the deprecation notice in the changelog by Friday?”. وقع القطع في منتصفها، فذهب الاسم إلى طلب والباقي إلى طلب آخر. سمع Whisper الاسم اليتيم سؤالًا، واخترع awesome. ليسدّ الفجوة في نهاية القطعة، وحوّل سطرًا واحدًا إلى ثلاثة. التوقيتات لا تزال صحيحة، وخطوة الملاحظات لا تزال تلتقط المهمة. ما تفقده هو الاسم، وهو ما تعتمد عليه الإسناد.

التقسيم على مدة ثابتة يقطع كلام أحدهم عند كل حدّ. عشرون ثانية قصيرة بما يكفي لتقع في منتصف جملة كل مرة تقريبًا؛ وعشر دقائق تجعل ذلك نادرًا لا مستحيلًا، وفي النهاية سيقع القطع على تلك الجملة الوحيدة التي تُسند العمل. التقسيم على الصمت يتجنّب المشكلة بشكل صحيح، ويحتاج إلى أداة قادرة على العثور على الفجوات، مثل ffmpeg أو pydub. اقسم فقط حين يستدعي الملف ذلك فعلًا.

لا يمكن تقطيع الصيغ المضغوطة بهذه الطريقة، إذ لا يمكنك قطع MP3 على حدود إطار (frame) بالمكتبة القياسية. استخدم ffmpeg معها:

ffmpeg -i meeting.mp3 -f segment -segment_time 600 -c copy chunk_%03d.mp3

تجميع القطع

def meeting_notes(path: str, attendees: list[str] | None = None) -> dict:
    model = "openai/whisper-large-v3"
    size_mb = os.path.getsize(path) / 1_000_000
    if path.endswith(".wav") and size_mb > 20:
        print(f"{size_mb:.0f} MB, splitting", file=sys.stderr)
        lines = transcribe_long(path, model)
    else:
        lines = timed_lines(transcribe(path, model, timestamps=True))
    print(f"{len(lines)} lines transcribed", file=sys.stderr)
    return write_notes(lines, attendees)


if __name__ == "__main__":
    recording = sys.argv[1] if len(sys.argv) > 1 else "standup.wav"
    roster = sys.argv[2:] or None
    print(json.dumps(meeting_notes(recording, roster), indent=2))

python notes.py standup.wav "Priya Raman" "Tomas Vidal" "Mei Lin"

الخطوات التالية

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

تحويل الكلام إلى نص

مرجع لنقطة نهاية التفريغ.

استخراج البيانات المُهيكلة من المستندات

الاستخراج نفسه المرتكز على المخطط، مطبَّقًا على الملفات.

الاستجابات المُهيكلة

كيف يقيّد json_schema إكمال المحادثة.

استنساخ الصوت

امنح الملخّص صوتًا خاصًّا به.

التوثيق

البدء

النص والدردشة

الصورة

الفيديو

الصوت

البحث وRAG

الوكلاء والتكاملات

أدوات البرمجة

أدوات الوكلاء

حِزم SDK والأُطر

ملاحظات الاجتماعات باستخدام تحويل الكلام إلى نص

الإعداد

1. تفريغ التسجيل

2. اطلب التوقيتات

3. استخراج الملاحظات

4. لا أحد مُصنَّف

5. أطول من طلب واحد

تجميع القطع

الخطوات التالية

تحويل الكلام إلى نص

استخراج البيانات المُهيكلة من المستندات

الاستجابات المُهيكلة

استنساخ الصوت

​الإعداد

​1. تفريغ التسجيل

​2. اطلب التوقيتات

​3. استخراج الملاحظات

​4. لا أحد مُصنَّف

​5. أطول من طلب واحد

​تجميع القطع

​الخطوات التالية

تحويل الكلام إلى نص

استخراج البيانات المُهيكلة من المستندات

الاستجابات المُهيكلة

استنساخ الصوت

الإعداد

1. تفريغ التسجيل

2. اطلب التوقيتات

3. استخراج الملاحظات

4. لا أحد مُصنَّف

5. أطول من طلب واحد

تجميع القطع

الخطوات التالية