wehub-resource-sync
295 lines
11 KiB
Python
295 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import os
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
from PIL import Image
|
|
|
|
from livekit.agents import (
|
|
DEFAULT_API_CONNECT_OPTIONS,
|
|
NOT_GIVEN,
|
|
APIConnectionError,
|
|
APIConnectOptions,
|
|
APIStatusError,
|
|
NotGivenOr,
|
|
utils,
|
|
)
|
|
|
|
from .log import logger
|
|
from .meeting.room import JoinMeetingResult
|
|
|
|
|
|
class LemonSliceException(Exception):
|
|
"""Exception for LemonSlice errors"""
|
|
|
|
|
|
DEFAULT_API_URL = "https://lemonslice.com/api/liveai/sessions"
|
|
|
|
|
|
class LemonSliceAPI:
|
|
def __init__(
|
|
self,
|
|
api_key: NotGivenOr[str] = NOT_GIVEN,
|
|
api_url: NotGivenOr[str] = NOT_GIVEN,
|
|
*,
|
|
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
|
session: aiohttp.ClientSession | None = None,
|
|
) -> None:
|
|
"""
|
|
Initializes the LemonSliceAPI client.
|
|
|
|
Args:
|
|
api_key: Your LemonSlice API key. If not provided, it is read from
|
|
the LEMONSLICE_API_KEY environment variable.
|
|
api_url: The base URL of the LemonSlice API.
|
|
conn_options: Connection options for the aiohttp session.
|
|
session: An optional existing aiohttp.ClientSession to use for requests.
|
|
"""
|
|
ls_api_key = api_key if utils.is_given(api_key) else os.getenv("LEMONSLICE_API_KEY")
|
|
if not ls_api_key:
|
|
raise LemonSliceException("LEMONSLICE_API_KEY must be set")
|
|
self._api_key = ls_api_key
|
|
|
|
self._api_url = api_url or DEFAULT_API_URL
|
|
self._conn_options = conn_options
|
|
self._session = session
|
|
self._owns_session = session is None
|
|
|
|
async def __aenter__(self) -> LemonSliceAPI:
|
|
if self._owns_session:
|
|
self._session = aiohttp.ClientSession()
|
|
return self
|
|
|
|
async def __aexit__(
|
|
self, exc_type: type | None, exc_val: Exception | None, exc_tb: Any
|
|
) -> None:
|
|
if self._owns_session and self._session and not self._session.closed:
|
|
await self._session.close()
|
|
|
|
async def start_agent_session(
|
|
self,
|
|
*,
|
|
livekit_url: str,
|
|
livekit_token: str,
|
|
livekit_session_id: str,
|
|
agent_id: NotGivenOr[str] = NOT_GIVEN,
|
|
agent_image_url: NotGivenOr[str] = NOT_GIVEN,
|
|
agent_image: NotGivenOr[Image.Image] = NOT_GIVEN,
|
|
agent_prompt: NotGivenOr[str] = NOT_GIVEN,
|
|
agent_idle_prompt: NotGivenOr[str] = NOT_GIVEN,
|
|
idle_timeout: NotGivenOr[int] = NOT_GIVEN,
|
|
extra_payload: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
|
|
) -> str:
|
|
"""
|
|
Initiates a new LemonSlice agent session.
|
|
|
|
Args:
|
|
livekit_url: The LiveKit Cloud server URL.
|
|
livekit_token: The LiveKit access token for the agent.
|
|
livekit_session_id: LiveKit room session ID (room SID).
|
|
agent_id: The ID of the LemonSlice agent to add to the session.
|
|
agent_image_url: The URL of the image to use as the agent's avatar.
|
|
agent_image: A PIL image to upload and use as the agent's avatar. Sent to
|
|
LemonSlice as a multipart image upload.
|
|
agent_prompt: A prompt that subtly influences the avatar's movements and expressions while responding.
|
|
agent_idle_prompt: A prompt that subtly influences the avatar's movements and expressions while idle.
|
|
idle_timeout: The idle timeout, in seconds.
|
|
extra_payload: Additional payload to include in the request.
|
|
|
|
Returns:
|
|
The unique session ID for the LemonSlice agent session.
|
|
"""
|
|
given_sources = [
|
|
source for source in (agent_id, agent_image_url, agent_image) if utils.is_given(source)
|
|
]
|
|
if len(given_sources) == 0:
|
|
raise LemonSliceException("Missing one of agent_id, agent_image_url or agent_image")
|
|
if len(given_sources) > 1:
|
|
raise LemonSliceException(
|
|
"Only one of agent_id, agent_image_url or agent_image can be provided"
|
|
)
|
|
|
|
payload: dict[str, Any] = {
|
|
"transport_type": "livekit",
|
|
"properties": {
|
|
"livekit_url": livekit_url,
|
|
"livekit_token": livekit_token,
|
|
"livekit_session_id": livekit_session_id,
|
|
},
|
|
}
|
|
|
|
image_bytes: bytes | None = None
|
|
|
|
if utils.is_given(agent_id):
|
|
payload["agent_id"] = agent_id
|
|
if utils.is_given(agent_image_url):
|
|
payload["agent_image_url"] = agent_image_url
|
|
if utils.is_given(agent_prompt):
|
|
payload["agent_prompt"] = agent_prompt
|
|
if utils.is_given(agent_idle_prompt):
|
|
payload["agent_idle_prompt"] = agent_idle_prompt
|
|
if utils.is_given(idle_timeout):
|
|
payload["idle_timeout"] = idle_timeout
|
|
if utils.is_given(extra_payload):
|
|
payload.update(extra_payload)
|
|
if utils.is_given(agent_image):
|
|
image_bytes = _encode_image(agent_image)
|
|
|
|
response_data = await self._post(payload, image_bytes=image_bytes)
|
|
session_id = response_data["session_id"]
|
|
logger.debug(f"LemonSlice Session ID = {session_id}")
|
|
return session_id # type: ignore
|
|
|
|
async def join_meeting(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
meeting_url: str,
|
|
livekit_url: str,
|
|
broadcast_token: str,
|
|
bot_name: NotGivenOr[str] = NOT_GIVEN,
|
|
) -> JoinMeetingResult:
|
|
"""Add an active avatar session to an external video meeting.
|
|
|
|
Supports Zoom, Google Meet, Microsoft Teams, and Webex.
|
|
|
|
Args:
|
|
session_id: LemonSlice agent session ID.
|
|
meeting_url: URL of the external meeting to join.
|
|
livekit_url: LiveKit server URL for the agent room.
|
|
broadcast_token: LiveKit token used to subscribe to avatar media.
|
|
bot_name: Optional display name for the bot in the meeting.
|
|
|
|
Returns:
|
|
JoinMeetingResult with relay WebSocket URL and meeting bot ID.
|
|
"""
|
|
payload: dict[str, Any] = {
|
|
"session_id": session_id,
|
|
"meeting_url": meeting_url,
|
|
"livekit_url": livekit_url,
|
|
"broadcast_token": broadcast_token,
|
|
}
|
|
if utils.is_given(bot_name) and bot_name:
|
|
payload["bot_name"] = bot_name
|
|
|
|
url = f"{self._api_url.rstrip('/')}/{session_id}/join-meeting"
|
|
data = await self._post(payload, url=url)
|
|
return JoinMeetingResult(
|
|
websocket_url=str(data["websocket_url"]),
|
|
meeting_bot_id=str(data["meeting_bot_id"]),
|
|
)
|
|
|
|
async def leave_meeting(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
meeting_bot_id: str,
|
|
) -> None:
|
|
"""Remove the avatar from an external meeting.
|
|
|
|
Args:
|
|
session_id: LemonSlice agent session ID.
|
|
meeting_bot_id: Meeting bot ID returned by join_meeting().
|
|
"""
|
|
url = f"{self._api_url.rstrip('/')}/{session_id}/leave-meeting"
|
|
await self._post(
|
|
{"meeting_bot_id": meeting_bot_id},
|
|
url=url,
|
|
)
|
|
|
|
async def _post(
|
|
self,
|
|
payload: dict[str, Any],
|
|
*,
|
|
url: str | None = None,
|
|
image_bytes: bytes | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Make a POST request to the LemonSlice API with retry logic.
|
|
|
|
Args:
|
|
payload: JSON payload for the request.
|
|
url: Optional URL override.
|
|
image_bytes: Optional PNG-encoded image.
|
|
|
|
Returns:
|
|
Response data as a dictionary
|
|
|
|
Raises:
|
|
APIConnectionError: If the request fails after all retries
|
|
"""
|
|
session = self._session or aiohttp.ClientSession()
|
|
try:
|
|
for i in range(self._conn_options.max_retry + 1):
|
|
try:
|
|
headers = {"X-API-Key": self._api_key}
|
|
request_kwargs: dict[str, Any]
|
|
if image_bytes is not None:
|
|
form = aiohttp.FormData()
|
|
form.add_field(
|
|
"payload", json.dumps(payload), content_type="application/json"
|
|
)
|
|
# Upload the image using multipart
|
|
form.add_field(
|
|
"image",
|
|
image_bytes,
|
|
filename="image.png",
|
|
content_type="image/png",
|
|
)
|
|
request_kwargs = {"data": form}
|
|
else:
|
|
headers["Content-Type"] = "application/json"
|
|
request_kwargs = {"json": payload}
|
|
|
|
async with session.post(
|
|
url or self._api_url,
|
|
headers=headers,
|
|
timeout=aiohttp.ClientTimeout(
|
|
total=60.0,
|
|
sock_connect=self._conn_options.timeout,
|
|
),
|
|
**request_kwargs,
|
|
) as response:
|
|
if not response.ok:
|
|
text = await response.text()
|
|
raise APIStatusError(
|
|
"LemonSlice Server returned an error",
|
|
status_code=response.status,
|
|
body=text,
|
|
)
|
|
return await response.json() # type: ignore
|
|
except Exception as e:
|
|
if isinstance(e, APIStatusError):
|
|
logger.error(
|
|
"LemonSlice API returned an error",
|
|
extra={
|
|
"status_code": e.status_code,
|
|
"body": e.body,
|
|
},
|
|
)
|
|
if not e.retryable:
|
|
raise e
|
|
elif isinstance(e, APIConnectionError):
|
|
logger.warning("failed to call LemonSlice api", extra={"error": str(e)})
|
|
else:
|
|
logger.exception("failed to call lemonslice api")
|
|
|
|
if i < self._conn_options.max_retry:
|
|
await asyncio.sleep(self._conn_options._interval_for_retry(i))
|
|
finally:
|
|
if not self._session: # if we created the session, we close it
|
|
await session.close()
|
|
|
|
raise APIConnectionError("Failed to call LemonSlice API after all retries")
|
|
|
|
|
|
def _encode_image(image: Image.Image) -> bytes:
|
|
"""Encode a PIL image as PNG bytes for a multipart upload."""
|
|
buffer = io.BytesIO()
|
|
image.save(buffer, format="PNG")
|
|
return buffer.getvalue()
|