client-events: add RemoteSession

livekit-agents/livekit/agents/evals/evaluation.py

@@ -73,6 +73,7 @@ def none_failed(self) -> bool:
         """True if no judgments explicitly failed. Maybes are allowed."""
         return not any(j.failed for j in self.judgments.values())
 
+
 class JudgeGroup:
     """A group of judges that evaluate conversations together.
 

livekit-agents/livekit/agents/evals/judge.py

@@ -4,12 +4,12 @@
 from typing import Any, Literal
 
 from ..llm import LLM, ChatContext, function_tool, utils as llm_utils
-from ..types import NOT_GIVEN, NotGivenOr
 from ..log import logger
 
 Verdict = Literal["pass", "fail", "maybe"]
 """The verdict of a judgment: pass, fail, or maybe (uncertain)."""
 
+
 @dataclass
 class JudgmentResult:
     verdict: Verdict
@@ -86,8 +86,7 @@ def _has_handoffs(chat_ctx: ChatContext) -> bool:
     Excludes initial agent assignments (where old_agent_id is None).
     """
     return any(
-        item.type == "agent_handoff" and item.old_agent_id is not None
-        for item in chat_ctx.items
+        item.type == "agent_handoff" and item.old_agent_id is not None for item in chat_ctx.items
     )
 
 
@@ -147,9 +146,7 @@ async def submit_verdict(verdict: Verdict, reasoning: str) -> tuple[Verdict, str
 
 
 class Judge:
-    def __init__(
-        self, *, llm: LLM | None = None, instructions: str, name: str = "custom"
-    ) -> None:
+    def __init__(self, *, llm: LLM | None = None, instructions: str, name: str = "custom") -> None:
         self._llm = llm
         self._instructions = instructions
         self._name = name

livekit-agents/livekit/agents/job.py

@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import asyncio
-import contextlib
 import contextvars
 import functools
 import inspect
@@ -27,19 +26,18 @@
 from dataclasses import dataclass
 from enum import Enum, unique
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, cast
+from typing import TYPE_CHECKING, Any, Callable
 from urllib.parse import urlparse
 
 import aiohttp
-from opentelemetry import trace
 
 from livekit import api, rtc
 from livekit.api.access_token import Claims
 from livekit.protocol import agent, models
 
 from .log import logger
 from .observability import Tagger
-from .telemetry import _upload_session_report, trace_types, tracer
+from .telemetry import _upload_session_report
 from .telemetry.traces import _setup_cloud_tracer, _shutdown_telemetry
 from .types import NotGivenOr
 from .utils import http_context, is_given, wait_for_participant
@@ -741,4 +739,3 @@ async def accept(
 class _JobShutdownInfo:
     user_initiated: bool
     reason: str
-

livekit-agents/livekit/agents/llm/chat_context.py

@@ -223,6 +223,7 @@ class AgentConfigUpdate(BaseModel):
     _tools: list[Tool] = PrivateAttr(default_factory=list)
     """Full tool definitions (in-memory only, not serialized)."""
 
+
 ChatItem = Annotated[
     Union[ChatMessage, FunctionCall, FunctionCallOutput, AgentHandoff, AgentConfigUpdate],
     Field(discriminator="type"),

livekit-agents/livekit/agents/observability.py

@@ -94,8 +94,10 @@ def _evaluation(self, result: EvaluationResult) -> None:
         """
         for name, judgment in result.judgments.items():
             self._tags.add(f"lk.judge.{name}:{judgment.verdict}")
-            self._evaluation_results.append({
-                "name": name,
-                "verdict": judgment.verdict,
-                "reasoning": judgment.reasoning,
-            })
+            self._evaluation_results.append(
+                {
+                    "name": name,
+                    "verdict": judgment.verdict,
+                    "reasoning": judgment.reasoning,
+                }
+            )

livekit-agents/livekit/agents/telemetry/traces.py

@@ -382,7 +382,6 @@ def _log(
             severity_text=severity_text,
         )
 
-
     eval_logger = _get_logger("evaluations")
     if tagger.evaluations:
         for evaluation in tagger.evaluations:

livekit-agents/livekit/agents/types.py

@@ -24,6 +24,30 @@
 
 TOPIC_CHAT = "lk.chat"
 TOPIC_TRANSCRIPTION = "lk.transcription"
+TOPIC_CLIENT_EVENTS = "lk.agent.events"
+"""
+Topic for streaming agent events to room participants.
+"""
+
+RPC_GET_SESSION_STATE = "lk.agent.get_session_state"
+"""
+RPC method to get the current session state.
+"""
+
+RPC_GET_CHAT_HISTORY = "lk.agent.get_chat_history"
+"""
+RPC method to get the agent<>user conversation turns.
+"""
+
+RPC_GET_AGENT_INFO = "lk.agent.get_agent_info"
+"""
+RPC method to get information about the current agent.
+"""
+
+RPC_SEND_MESSAGE = "lk.agent.send_message"
+"""
+RPC method to send a message and get the agent's response.
+"""
 
 USERDATA_TIMED_TRANSCRIPT = "lk.timed_transcripts"
 """

livekit-agents/livekit/agents/voice/agent_session.py

@@ -41,6 +41,7 @@
 from .agent import Agent
 from .agent_activity import AgentActivity
 from .audio_recognition import TurnDetectionMode
+from .client_events import ClientEventsHandler
 from .events import (
     AgentEvent,
     AgentState,
@@ -334,6 +335,7 @@ def __init__(
         # used to keep a reference to the room io
         self._room_io: room_io.RoomIO | None = None
         self._recorder_io: RecorderIO | None = None
+        self._client_events_handler: ClientEventsHandler | None = None
 
         self._agent: Agent | None = None
         self._activity: AgentActivity | None = None
@@ -524,6 +526,7 @@ async def start(
             self._recorded_events = []
             self._room_io = None
             self._recorder_io = None
+            self._client_events_handler = None
 
             self._closing = False
             self._root_span_context = otel_context.get_current()
@@ -576,6 +579,19 @@ async def start(
                 self._room_io = room_io.RoomIO(room=room, agent_session=self, options=room_options)
                 await self._room_io.start()
 
+                # Initialize the client events handler for exposing session state to clients
+                self._client_events_handler = ClientEventsHandler(
+                    session=self,
+                    room_io=self._room_io,
+                )
+
+                # Register text input handler if configured
+                text_input_opts = room_options.get_text_input_options()
+                if text_input_opts:
+                    self._client_events_handler.register_text_input(text_input_opts.text_input_cb)
+
+                await self._client_events_handler.start()
+
             if job_ctx:
                 # these aren't relevant during eval mode, as they require job context and/or room_io
                 if self.input.audio and self.output.audio:
@@ -826,6 +842,11 @@ async def _aclose_impl(
             self._tts_error_counts = 0
             self._root_span_context = None
 
+            # close client events handler before room io
+            if self._client_events_handler:
+                await self._client_events_handler.aclose()
+                self._client_events_handler = None
+
             # close room io after close event is emitted
             if self._room_io:
                 await self._room_io.aclose()