Khái niệm cốt lõi của LiveKit

LiveKit hoạt động dựa trên ba khái niệm chính:

• Room (Phòng): Một không gian ảo nơi các thành viên có thể kết nối và tương tác.
• Participant (Thành viên): Đại diện cho mỗi người dùng tham gia vào một phòng.
• Track (Luồng): Đại diện cho các loại dữ liệu truyền tải, chẳng hạn như âm thanh, video hoặc luồng màn hình.

Thao tác Client

Tham gia phòng

Client có thể tham gia phòng theo hai cách:

• Tạo thủ công: Cấp quyền cho người dùng tạo phòng. Khi người dùng tham gia, họ sẽ tạo một phòng mới.
• Tham gia tự động: Người dùng cung cấp URL WebSocket và token để tham gia vào một phòng cụ thể.

Ví dụ tham gia phòng:

room = LiveKit.create(appContext = applicationContext)
room.connect(wsUrl, token)
    

Rời khỏi phòng

Để rời khỏi phòng, gọi phương thức Room.disconnect(). Nếu ứng dụng đóng đột ngột mà không thông báo cho LiveKit, người dùng vẫn sẽ hiển thị trong phòng khoảng 15 giây. Trên nền tảng Swift, Room.disconnect được gọi tự động khi ứng dụng thoát.

Gửi tin nhắn

Client có thể gửi tin nhắn tùy ý tới các thành viên khác trong phòng thông qua API LocalParticipant.publishData. Tin nhắn được gửi qua kênh dữ liệu WebRTC đến SFU (Selective Forwarding Unit), sau đó LiveKit sẽ chuyển tiếp tới các thành viên đích.

Để gửi tin nhắn đến một người dùng cụ thể, bạn có thể chỉ định destinationIdentities.

Ví dụ gửi và nhận tin nhắn:

// Gửi tin nhắn
coroutineScope.launch {
  val data: ByteArray = // Dữ liệu cần gửi

  // Gửi tin nhắn không đảm bảo (LOSSY) tới tất cả mọi người
  // LOSSY: Ưu tiên tốc độ, không đảm bảo thứ tự hoặc khả năng gửi. Lý tưởng cho cập nhật thời gian thực.
  room.localParticipant.publishData(data, DataPublishReliability.LOSSY)

  // Gửi tin nhắn có đảm bảo (RELIABLE) tới các thành viên cụ thể
  // RELIABLE: Đảm bảo gửi ít nhất một lần (tối đa 3 lần thử lại) và đảm bảo thứ tự. Phù hợp cho tin nhắn quan trọng.
  val targetIdentities = listOf(
    Participant.Identity("alice"),
    Participant.Identity("bob"),
  )
  room.localParticipant.publishData(data, DataPublishReliability.RELIABLE, targetIdentities)
}

// Xử lý tin nhắn nhận được
coroutineScope.launch {
  room.events.collect { event ->
    if (event is RoomEvent.DataReceived) {
        // Xử lý dữ liệu nhận được
    }
  }
}
    

Giới hạn kích thước tin nhắn

Do giới hạn của giao thức SCTP, việc gửi tin nhắn qua kênh dữ liệu lớn hơn 16 KiB không thực tế. LiveKit khuyến nghị giữ kích thước tin nhắn dưới 15 KiB.

Topic tin nhắn

Tin nhắn có thể được gán một topic, cho phép bên nhận lọc và chỉ xử lý các tin nhắn quan tâm.

Truyền luồng dữ liệu

LiveKit hỗ trợ mặc định các luồng camera, micro và chia sẻ màn hình. Bạn cũng có thể cấu hình để truyền các luồng tùy chỉnh.

Luồng âm thanh và video

// Bật camera
room.localParticipant.setCameraEnabled(true)

// Bật micro
room.localParticipant.setMicrophoneEnabled(true)
    

Luồng chia sẻ màn hình

// Khởi tạo trình khởi chạy ý định chụp màn hình
val screenCaptureIntentLauncher = registerForActivityResult(
    ActivityResultContracts.StartActivityForResult()
) { result ->
    val resultCode = result.resultCode
    val data = result.data
    if (resultCode != Activity.RESULT_OK || data == null) {
        return@registerForActivityResult
    }
    lifecycleScope.launch {
        room.localParticipant.setScreenShareEnabled(true, data)
    }
}

// Khi cần bật chia sẻ màn hình
val mediaProjectionManager = getSystemService(MEDIA_PROJECTION_SERVICE) as MediaProjectionManager
screenCaptureIntentLauncher.launch(mediaProjectionManager.createScreenCaptureIntent())
    

Cấu hình luồng tùy chỉnh

Tùy chọn 1: Thiết lập mặc định cho phòng

val options = RoomOptions(
  audioTrackCaptureDefaults = LocalAudioTrackOptions(
    noiseSuppression = true,
    echoCancellation = true,
    autoGainControl = true,
    highPassFilter = true,
    typingNoiseDetection = true,
  ),
  videoTrackCaptureDefaults = LocalVideoTrackOptions(
    deviceId = "",
    position = CameraPosition.FRONT,
    captureParams = VideoPreset169.H1080.capture,
  ),
  audioTrackPublishDefaults = AudioTrackPublishDefaults(
    audioBitrate = 20_000,
    dtx = true,
  ),
  videoTrackPublishDefaults = VideoTrackPublishDefaults(
    videoEncoding = VideoPreset169.H1080.encoding,
  )
)
var room = LiveKit.create(
  ...
  roomOptions = options,
)
    

Tùy chọn 2: Tạo luồng thủ công

val localParticipant = room.localParticipant
val audioTrack = localParticipant.createAudioTrack("audio")
localParticipant.publishAudioTrack(audioTrack)

val videoTrack = localParticipant.createVideoTrack("video", LocalVideoTrackOptions(
  CameraPosition.FRONT,
  VideoPreset169.H1080.capture
))
localParticipant.publishVideoTrack(videoTrack)
    

Theo dõi luồng dữ liệu

Theo mặc định, khi một người dùng tham gia phòng, họ sẽ tự động theo dõi tất cả các luồng dữ liệu.

coroutineScope.launch {
  room.events.collect { event ->
    when(event) {
      is RoomEvent.TrackSubscribed -> {
        // Luồng âm thanh được phát tự động.
        val videoTrack = event.track as? VideoTrack ?: return@collect
        videoTrack.addRenderer(videoRenderer)
      }
      else -> {}
    }
  }
}
    

Lắng nghe sự kiện

Các sự kiện trong LiveKit được chia thành sự kiện phòng (Room Event) và sự kiện thành viên (Participant Event).

SỰ KIỆN	MÔ TẢ	ROOM EVENT	PARTICIPANT EVENT
ParticipantConnected	Một thành viên từ xa tham gia phòng sau thành viên cục bộ.	✅
ParticipantDisconnected	Một thành viên từ xa rời khỏi phòng.	✅
Reconnecting	Kết nối với máy chủ bị gián đoạn và đang cố gắng kết nối lại.	✅
Reconnected	Kết nối lại thành công.	✅
Disconnected	Ngắt kết nối khỏi phòng do phòng đóng hoặc lỗi không thể phục hồi.	✅
TrackPublished	Một luồng mới được xuất bản trong phòng sau khi thành viên cục bộ tham gia.	✅	✅
TrackUnpublished	Một thành viên từ xa đã hủy xuất bản một luồng.	✅	✅
TrackSubscribed	Thành viên cục bộ đã đăng ký theo dõi một luồng.	✅	✅
TrackUnsubscribed	Một luồng đã đăng ký trước đó đã bị hủy đăng ký.	✅	✅
TrackMuted	Một luồng bị tắt tiếng (áp dụng cho cả luồng cục bộ và từ xa).	✅	✅
TrackUnmuted	Một luồng được bật tiếng trở lại (áp dụng cho cả luồng cục bộ và từ xa).	✅	✅
LocalTrackPublished	Một luồng cục bộ đã được xuất bản thành công.	✅	✅
LocalTrackUnpublished	Một luồng cục bộ đã bị hủy xuất bản.	✅	✅
ActiveSpeakersChanged	Danh sách những người đang nói tích cực hiện tại đã thay đổi.	✅
IsSpeakingChanged	Trạng thái đang nói của thành viên hiện tại đã thay đổi.		✅
ConnectionQualityChanged	Chất lượng kết nối của một thành viên đã thay đổi.	✅	✅
ParticipantMetadataChanged	Siêu dữ liệu của thành viên đã được cập nhật thông qua API máy chủ.	✅	✅
RoomMetadataChanged	Siêu dữ liệu liên quan đến phòng đã thay đổi.	✅
DataReceived	Dữ liệu nhận được từ một thành viên khác hoặc máy chủ.	✅	✅
TrackStreamStateChanged	Chỉ báo liệu một luồng đã đăng ký có bị tạm dừng do băng thông hay không.	✅	✅
TrackSubscriptionPermissionChanged	Một trong các luồng đã đăng ký đã thay đổi quyền cấp luồng cho thành viên hiện tại.	✅	✅
ParticipantPermissionsChanged	Khi quyền của thành viên hiện tại đã thay đổi.	✅	✅

Thao tác Server

Tạo token người dùng

Để tạo token, bạn cần API_KEY và API_SECRET từ dịch vụ LiveKit. Token này là một JWT (JSON Web Token) được sử dụng để xác thực và ủy quyền cho người dùng.

Thông tin người dùng (identity, name, room) có thể được nhúng vào token.

Ví dụ tạo token bằng Python:

import os
from livekit import api
from flask import Flask

app = Flask(__name__)

@app.route('/getToken')
def getToken():
 token = api.AccessToken(os.getenv('LIVEKIT_API_KEY'), os.getenv('LIVEKIT_API_SECRET')) \
 .with_identity("identity") \
 .with_name("my name") \
 .with_grants(api.VideoGrants(
 room_join=True,
 room="my-room",
 ))
 return token.to_jwt()
    

Bạn cũng có thể tạo token nhanh chóng bằng CLI của LiveKit:

livekit-cli token create --api-key devkey --api-secret secret --join --room test_room --identity test_user --valid-for 24h
    

Thuộc tính của token

Token JWT chứa các thông tin như định danh người dùng, tên phòng, chức năng và quyền hạn. Quyền hạn trong phòng được chỉ định trong trường video của token đã giải mã.

TRƯỜNG	KIỂU	MÔ TẢ
roomCreate	bool	Quyền tạo hoặc xóa phòng.
roomList	bool	Quyền liệt kê các phòng khả dụng.
roomJoin	bool	Quyền tham gia phòng.
roomAdmin	bool	Quyền quản lý phòng.
roomRecord	bool	Quyền sử dụng dịch vụ Egress.
ingressAdmin	bool	Quyền sử dụng dịch vụ Ingress.
room	string	Tên phòng (bắt buộc nếu join hoặc admin được đặt).
canPublish	bool	Cho phép thành viên xuất bản luồng.
canPublishData	bool	Cho phép thành viên gửi dữ liệu trong phòng.
canPublishSources	string[]	Khi được đặt, chỉ cho phép xuất bản các nguồn được liệt kê (camera, microphone, screen_share, screen_share_audio).
canSubscribe	bool	Cho phép thành viên đăng ký theo dõi luồng.
canUpdateOwnMetadata	bool	Cho phép thành viên cập nhật siêu dữ liệu của chính họ.
hidden	bool	Ẩn thành viên khỏi những người khác trong phòng.
kind	string	Loại thành viên (standard, ingress, egress, sip, agent). Trường này thường được đặt bởi nội bộ LiveKit.

Thao tác ngắt kết nối phiên

Khi người dùng rời phòng, phiên của họ sẽ kết thúc. Bạn có thể sử dụng hàm callback add_shutdown_callback để xử lý các tác vụ sau đó, ví dụ: gửi sự kiện kết thúc trò chuyện.

async def entrypoint(ctx: JobContext):
    async def my_shutdown_hook():
        # Lưu trạng thái người dùng
        ...
    ctx.add_shutdown_callback(my_shutdown_hook)
    

Thao tác Agent

Thiết lập nút dịch vụ Agent

LiveKit Agent SDK hiện chỉ hỗ trợ Python. Tài liệu chi tiết tại: LiveKit Agents Quickstart.

Ví dụ demo từ LiveKit:

import asyncio

from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, cli, llm
from livekit.agents.voice_assistant import VoiceAssistant
from livekit.plugins import deepgram, openai, silero

# Đây là điểm bắt đầu cho agent.
async def entrypoint(ctx: JobContext):
    # Tạo ngữ cảnh trò chuyện ban đầu với một lời nhắc hệ thống
    initial_ctx = llm.ChatContext().append(
        role="system",
        text=(
            "You are a voice assistant created by LiveKit. Your interface with users will be voice. "
            "You should use short and concise responses, and avoiding usage of unpronouncable punctuation."
        ),
    )

    # Kết nối đến phòng LiveKit
    # Chỉ định rằng agent sẽ chỉ đăng ký theo dõi các luồng âm thanh
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    # VoiceAssistant là một lớp tạo ra một agent AI đàm thoại hoàn chỉnh.
    assistant = VoiceAssistant(
        vad=silero.VAD.load(),
        stt=deepgram.STT(),
        llm=openai.LLM(),
        tts=openai.TTS(),
        chat_ctx=initial_ctx,
    )

    # Khởi chạy voice assistant với phòng LiveKit
    assistant.start(ctx.room)

    await asyncio.sleep(1)

    # Chào người dùng với một thông điệp ban đầu
    await assistant.say("Hey, how can I help you today?", allow_interruptions=True)

if __name__ == "__main__":
    # Khởi tạo worker với hàm entrypoint
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
    

Vòng đời của Agent

1. Khi chương trình worker khởi động, nó sẽ kết nối đến máy chủ LiveKit qua WebSocket và đăng ký làm worker. Mỗi worker có thể có nhiều tiến trình con (Agent) để xử lý các yêu cầu.
2. Khi người dùng tham gia phòng, máy chủ LiveKit sẽ chọn một worker thông qua cân bằng tải để phục vụ người dùng.
3. Tiến trình con của Agent xử lý tin nhắn từ người dùng và đưa ra phản hồi.
4. Khi người dùng rời phòng, phòng sẽ đóng lại và kết nối với agent bị ngắt.

Quy trình thực thi nội bộ của Agent

Agent xử lý yêu cầu qua các giai đoạn sau:

• Request handler: Xác định xem agent có thể xử lý yêu cầu hay không. Nếu không, LiveKit sẽ chuyển nhiệm vụ cho worker khác.
• Entrypoint: Các thao tác khởi tạo được thực hiện trước khi agent tham gia phòng.
• Prewarm function: Được gọi khi tiến trình agent khởi động, cho phép thực hiện các tác vụ tốn thời gian như tải mô hình.

Loại Worker

opts = WorkerOptions(
    ...
    # Nếu không chỉ định, mặc định là JobType.JT_ROOM
    worker_type=JobType.JT_ROOM,
)
    

Enum JobType có hai tùy chọn:

• JT_ROOM: Một phiên bản agent mới sẽ được tạo cho mỗi phòng.
• JT_PUBLISHER: Một phiên bản agent mới sẽ được tạo cho mỗi thành viên trong phòng.

Xử lý luồng âm thanh

@ctx.room.on("track_subscribed")
def on_track_subscribed(
    track: rtc.Track,
    publication: rtc.TrackPublication,
    participant: rtc.RemoteParticipant,
):
    # Theo dõi luồng âm thanh
    if track.kind == rtc.TrackKind.KIND_AUDIO:
        audio_stream = rtc.AudioStream(track)
        async for event in audio_stream:
            do_something(event.frame)
    

Xuất bản luồng âm thanh

Để xuất bản âm thanh, luồng cần được chia thành các khung âm thanh có độ dài cố định. Bộ đệm nội bộ giữ một hàng đợi âm thanh dài 50ms để gửi theo thời gian thực. Phương thức capture_frame để gửi khung mới là phương thức chặn, nó sẽ chặn cho đến khi bộ đệm nhận đủ toàn bộ khung. Điều này giúp xử lý các gián đoạn dễ dàng hơn.

Để xuất bản một luồng âm thanh, bạn cần xác định trước tần số lấy mẫu, số kênh và độ dài mỗi khung (số mẫu). Ví dụ sau đây truyền một sóng sin có biên độ không đổi trong các khung dài 10ms ở tần số 48kHz:

SAMPLE_RATE = 48000
NUM_CHANNELS = 1 # âm thanh đơn kênh
AMPLITUDE = 2 ** 8 - 1
SAMPLES_PER_CHANNEL = 480 # 10ms ở 48kHz

async def entrypoint(ctx: JobContext):
    await ctx.connect()

    source = rtc.AudioSource(SAMPLE_RATE, NUM_CHANNELS)
    track = rtc.LocalAudioTrack.create_audio_track("example-track", source)
    # Vì agent là một thành viên, I/O âm thanh của nó là "micro" của nó
    options = rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE)
    # ctx.agent là bí danh cho ctx.room.local_participant
    publication = await ctx.agent.publish_track(track, options)

    frequency = 440
    async def _sinewave():
        audio_frame = rtc.AudioFrame.create(SAMPLE_RATE, NUM_CHANNELS, SAMPLES_PER_CHANNEL)
        audio_data = np.frombuffer(audio_frame.data, dtype=np.int16)

        time = np.arange(SAMPLES_PER_CHANNEL) / SAMPLE_RATE
        total_samples = 0
        while True:
            time = (total_samples + np.arange(SAMPLES_PER_CHANNEL)) / SAMPLE_RATE
            sinewave = (AMPLITUDE * np.sin(2 * np.pi * frequency * time)).astype(np.int16)
            np.copyto(audio_data, sinewave)

            # gửi khung này đến luồng
            await source.capture_frame(audio_frame)
            total_samples += SAMPLES_PER_CHANNEL
    

Xử lý tin nhắn văn bản

Lắng nghe sự kiện data_received để xử lý tin nhắn từ người dùng. Sử dụng publish_data() để gửi tin nhắn.

@room.on("data_received")
def on_data_received(data: rtc.DataPacket):
  logging.info("received data from %s: %s", data.participant.identity, data.data)

# Dữ liệu dạng chuỗi sẽ được mã hóa thành byte với UTF-8
await room.local_participant.publish_data("my payload",
                reliable=True,
                destination_identities=["identity1", "identity2"],
                topic="topic1")