session.history.to_dict()

timestamp

to_dict()

timestamp

import asyncio
import logging
from collections.abc import AsyncGenerator, AsyncIterable

from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
from livekit.agents.voice.agent import ModelSettings
from <http://livekit.agents.voice.io|livekit.agents.voice.io> import TimedString
from livekit.plugins import deepgram, openai, cartesia, silero

logger = logging.getLogger("my-worker")
logger.setLevel(<http://logging.INFO|logging.INFO>)

class MyAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful assistant.")
        self._closing_task: asyncio.Task[None] | None = None

    async def transcription_node(
        self, text: AsyncIterable[str | TimedString], model_settings: ModelSettings
    ) -> AsyncGenerator[str | TimedString, None]:
        async for chunk in text:
            if isinstance(chunk, TimedString):
                # Here, you’ll get sentence-level timestamps for TTS providers without word-level timing
                <http://logger.info|logger.info>(f"TimedString: '{chunk}' ({chunk.start_time} - {chunk.end_time})")
            yield chunk

async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=deepgram.STT(),
        llm=openai.LLM(),
        tts=cartesia.TTS(),  # Feel free to use any TTS provider for sentence-level timing
        vad=silero.VAD.load(),
        use_tts_aligned_transcript=True,  # This enables TTS-aligned transcripts
    )

    await session.start(agent=MyAgent(), room=ctx.room)
    await session.generate_reply(instructions="say hello to the user")

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

use_tts_aligned_transcript=True

AgentSession

transcription_node

TimedString

start_time

end_time

use_tts_aligned_transcript=True

AgentSession

session = AgentSession(
    stt=deepgram.STT(),
    llm=openai.LLM(),
    tts=cartesia.TTS(),  # or another TTS provider
    vad=silero.VAD.load(),
    use_tts_aligned_transcript=True,
)

transcription_node

TimedString

start_time

end_time

async def transcription_node(
    self, 
    text: AsyncIterable[str | TimedString], 
    model_settings: ModelSettings
) -> AsyncGenerator[str | TimedString, None]:
    async for chunk in text:
        if isinstance(chunk, TimedString):
            # This contains start_time and end_time in seconds
            <http://logger.info|logger.info>(f"TimedString: '{chunk}' ({chunk.start_time} - {chunk.end_time})")
        yield chunk

openai.realtime.RealtimeModel