Module livekit.plugins.bithuman

BitHuman plugin for LiveKit Agents

See https://docs.livekit.io/agents/integrations/avatar/bithuman/ for more information.

Classes

class AvatarSession (*,
model_path: NotGivenOr[str | None] = NOT_GIVEN,
api_url: NotGivenOr[str] = NOT_GIVEN,
api_secret: NotGivenOr[str] = NOT_GIVEN,
api_token: NotGivenOr[str] = NOT_GIVEN,
runtime: NotGivenOr[AsyncBithuman | None] = NOT_GIVEN)

Expand source code

class AvatarSession:
    """A Beyond Presence avatar session"""

    def __init__(
        self,
        *,
        model_path: NotGivenOr[str | None] = NOT_GIVEN,
        api_url: NotGivenOr[str] = NOT_GIVEN,
        api_secret: NotGivenOr[str] = NOT_GIVEN,
        api_token: NotGivenOr[str] = NOT_GIVEN,
        runtime: NotGivenOr[AsyncBithuman | None] = NOT_GIVEN,
    ) -> None:
        self._api_url = api_url or (os.getenv("BITHUMAN_API_URL") or NOT_GIVEN)
        self._api_secret = api_secret or (os.getenv("BITHUMAN_API_SECRET") or NOT_GIVEN)
        self._api_token = api_token or (os.getenv("BITHUMAN_API_TOKEN") or NOT_GIVEN)
        self._model_path = model_path or (os.getenv("BITHUMAN_MODEL_PATH") or NOT_GIVEN)

        if self._api_secret is None and self._api_token is None:
            raise BitHumanException("BITHUMAN_API_SECRET or BITHUMAN_API_TOKEN must be set")
        if self._model_path is None:
            raise BitHumanException("BITHUMAN_MODEL_PATH must be set")

        self._avatar_runner: AvatarRunner | None = None
        self._runtime = runtime

    async def start(self, agent_session: AgentSession, room: rtc.Room) -> None:
        if self._runtime:
            runtime = self._runtime
            await runtime._initialize_token()  # refresh the token
        else:
            kwargs = {
                "model_path": self._model_path,
            }
            if self._api_secret:
                kwargs["api_secret"] = self._api_secret
            if self._api_token:
                kwargs["token"] = self._api_token
            if self._api_url:
                kwargs["api_url"] = self._api_url

            runtime = await AsyncBithuman.create(**kwargs)
            self._runtime = runtime
            await runtime.start()

        video_generator = BithumanGenerator(runtime)

        try:
            job_ctx = get_job_context()

            async def _on_shutdown() -> None:
                runtime.cleanup()

            job_ctx.add_shutdown_callback(_on_shutdown)
        except RuntimeError:
            pass

        output_width, output_height = video_generator.video_resolution
        avatar_options = AvatarOptions(
            video_width=output_width,
            video_height=output_height,
            video_fps=video_generator.video_fps,
            audio_sample_rate=video_generator.audio_sample_rate,
            audio_channels=1,
        )

        audio_buffer = QueueAudioOutput(sample_rate=runtime.settings.INPUT_SAMPLE_RATE)
        # create avatar runner
        self._avatar_runner = AvatarRunner(
            room=room,
            video_gen=video_generator,
            audio_recv=audio_buffer,
            options=avatar_options,
        )
        await self._avatar_runner.start()

        agent_session.output.audio = audio_buffer

    @property
    def runtime(self) -> AsyncBithuman:
        if self._runtime is None:
            raise BitHumanException("Runtime not initialized")
        return self._runtime

A Beyond Presence avatar session

Instance variables

prop runtime : AsyncBithuman

Expand source code

@property
def runtime(self) -> AsyncBithuman:
    if self._runtime is None:
        raise BitHumanException("Runtime not initialized")
    return self._runtime

Methods

async def start(self, agent_session: AgentSession, room: rtc.Room) ‑> None

Expand source code

async def start(self, agent_session: AgentSession, room: rtc.Room) -> None:
    if self._runtime:
        runtime = self._runtime
        await runtime._initialize_token()  # refresh the token
    else:
        kwargs = {
            "model_path": self._model_path,
        }
        if self._api_secret:
            kwargs["api_secret"] = self._api_secret
        if self._api_token:
            kwargs["token"] = self._api_token
        if self._api_url:
            kwargs["api_url"] = self._api_url

        runtime = await AsyncBithuman.create(**kwargs)
        self._runtime = runtime
        await runtime.start()

    video_generator = BithumanGenerator(runtime)

    try:
        job_ctx = get_job_context()

        async def _on_shutdown() -> None:
            runtime.cleanup()

        job_ctx.add_shutdown_callback(_on_shutdown)
    except RuntimeError:
        pass

    output_width, output_height = video_generator.video_resolution
    avatar_options = AvatarOptions(
        video_width=output_width,
        video_height=output_height,
        video_fps=video_generator.video_fps,
        audio_sample_rate=video_generator.audio_sample_rate,
        audio_channels=1,
    )

    audio_buffer = QueueAudioOutput(sample_rate=runtime.settings.INPUT_SAMPLE_RATE)
    # create avatar runner
    self._avatar_runner = AvatarRunner(
        room=room,
        video_gen=video_generator,
        audio_recv=audio_buffer,
        options=avatar_options,
    )
    await self._avatar_runner.start()

    agent_session.output.audio = audio_buffer

class BitHumanException (*args, **kwargs)

Expand source code

class BitHumanException(Exception):
    """Exception for BitHuman errors"""

Exception for BitHuman errors

Ancestors

• builtins.Exception
• builtins.BaseException