LiveKit Agents

version: "1.0"
 
mini-bcs-voice-livekit-env:
  services:
    - name: postgres
      config:
        SCHEMA_PATH: /agent/db/schema.sql
 
  actor:
    channels:
      - type: voice_ws
        url: ws://localhost:8008/voice
 
  agent:
    name: Mini BCS Voice (LiveKit)
    code_path: /agent
    # start.sh launches the three in-container processes: livekit-server (the
    # SFU), the Agents worker, and the voice_ws bridge that the actor talks to
    # on :8008. No veris-sandbox changes — the actor drives standard voice_ws.
    entry_point: bash /agent/start.sh
    environment:
      DATABASE_URL: postgresql://postgres:postgres@localhost:5432/veris
      LIVEKIT_URL: ws://localhost:7880
      LIVEKIT_API_KEY: devkey
      LIVEKIT_API_SECRET: secret
      # OPENAI_API_KEY drives the LiveKit OpenAI Realtime plugin. Inject once:
      #   veris env vars set OPENAI_API_KEY=sk-... --secret

from livekit.agents import Agent, AgentServer, AgentSession, JobContext, RunContext, cli
from livekit.agents.llm import ToolError, function_tool
from livekit.plugins import openai
 
# load_fnc always reports 0 so this dedicated, one-call-per-pod worker never
# self-throttles. By default a `start` (prod-mode) worker uses a CPU-based load
# function with a 0.7 threshold; when the SFU + worker + bridge + Realtime
# session share one CPU-bound sandbox pod under concurrent cluster load, that
# threshold trips and the SFU reports "no workers with sufficient capacity",
# so the agent never joins the room and the actor sees callee_no_answer.
# (dev mode defaults the threshold to inf for exactly this reason.)
server = AgentServer(load_fnc=lambda *_: 0.0)
 
 
@server.rtc_session()
async def entrypoint(ctx: JobContext) -> None:
    ctx.log_context_fields = {"room": ctx.room.name}
    logger.info("[entrypoint] joining room=%s", ctx.room.name)
 
    session = AgentSession(
        llm=openai.realtime.RealtimeModel(
            voice=os.environ.get("REALTIME_VOICE", "alloy"),
            model=os.environ.get("REALTIME_MODEL", "gpt-realtime"),
        ),
    )
    await session.start(agent=RileyAgent(), room=ctx.room)
    logger.info("[entrypoint] session started room=%s", ctx.room.name)
 
 
if __name__ == "__main__":
    cli.run_app(server)

    def __init__(self) -> None:
        super().__init__(instructions=AGENT_PROMPT)
        self._api = BCSAPI()
 
    async def on_enter(self) -> None:
        await self.session.generate_reply(
            instructions="Greet the caller as Riley from Acme Bank's credit card team and ask how you can help, in one short sentence."
        )

# 2. Agents worker — registers with the SFU so it can be dispatched. Capture
# its log so we can gate on *actual* registration, and mirror it to stdout so
# it still lands in agent.log for debugging.
uv run --no-sync python -m app.agent start > /tmp/worker.log 2>&1 &
WK_PID=$!
tail -f /tmp/worker.log 2>/dev/null &
TAIL_PID=$!
 
# Gate the bridge on the worker actually registering with the SFU — NOT a fixed
# sleep. Under concurrent cluster load the worker can take 10s+ to register; if
# the bridge accepts a caller before then, the room is created before the worker
# can be auto-dispatched into it, so the agent never joins and the actor reports
# callee_no_answer. Wait (up to 60s) for the "registered worker" log line.
echo "[start] waiting for the agent worker to register with the SFU..."
for _ in $(seq 1 120); do
  grep -q "registered worker" /tmp/worker.log 2>/dev/null && { echo "[start] worker registered — bringing up voice_ws bridge"; break; }
  kill -0 "$WK_PID" 2>/dev/null || { echo "[start] worker exited before registering" >&2; break; }
  sleep 0.5
done
 
# 3. voice_ws bridge / web server.
uv run --no-sync uvicorn app.web:app --host 0.0.0.0 --port "$PORT" &
WEB_PID=$!

class RileyAgent(Agent):
    # A read-only lookup: one success path, reported once.
    @function_tool()
    async def display_card_info_by_last4(
        self, context: RunContext, last4: str
    ) -> dict:
        """Find a card by the last 4 digits and return its details. Returns {} if not found."""
        card = self._api.find_card_by_last4(last4)
        result = card.model_dump() if card else {}
        report_tool_call("display_card_info_by_last4", {"last4": last4}, result)
        return result
 
    # A mutating tool: validate, report on both paths, raise ToolError on failure.
    @function_tool()
    async def change_card_status(
        self, context: RunContext, card_id: str, new_status: str
    ) -> dict:
        """Update a card's status. A cancelled card cannot change status."""
        args = {"card_id": card_id, "new_status": new_status}
        try:
            card = self._api.update_card_status(card_id, CardStatus(new_status))
            result = card.model_dump() if card else {}
            report_tool_call("change_card_status", args, result)
            return result
        except ValueError as exc:
            report_tool_call("change_card_status", args, {"error": str(exc)})
            raise ToolError(str(exc))

_ENGINE_URL = os.environ.get("ENGINE_URL", "http://localhost:6100")
_SIMULATION_ID = os.environ.get("SIMULATION_ID")
 
 
def _emit_tool_event(name: str, args: dict, result: object) -> None:
    body = json.dumps(
        {
            "service": "agent",
            "event_type": "agent_tool_call",
            "data": {"name": name, "arguments": args, "result": result},
        },
        default=str,  # enums/datetimes — same handling as the tool result
    )
    try:
        httpx.post(
            f"{_ENGINE_URL}/simulations/{_SIMULATION_ID}/events",
            content=body,
            headers={"Content-Type": "application/json"},
            timeout=2.0,
        )
    except Exception as exc:
        logger.warning("[tool] could not report %s to engine: %s", name, exc)
 
 
def report_tool_call(name: str, args: dict, result: object) -> None:
    """Fire-and-forget report of a tool call to the Veris engine.
 
    No-op outside a simulation. Runs the blocking POST in a worker thread so it
    never stalls the realtime voice loop, even if the engine is slow/unreachable.
    """
    if not _SIMULATION_ID:
        return
    task = asyncio.create_task(asyncio.to_thread(_emit_tool_event, name, args, result))
    _report_tasks.add(task)
    task.add_done_callback(_report_tasks.discard)

veris env vars set OPENAI_API_KEY=sk-... --secret

ARG VERIS_BASE
FROM ${VERIS_BASE}
 
# livekit-server — the in-container SFU the agent worker and the /voice bridge
# both connect to on localhost:7880. Pinned to a fixed release for reproducible,
# CI-stable builds. Dropped in /usr/local/bin (on PATH; no sudo needed as root).
ARG LIVEKIT_SERVER_VERSION=1.12.0
USER root
RUN apt-get update \
    && apt-get install -y --no-install-recommends curl ca-certificates tar \
    && curl -fsSL "https://github.com/livekit/livekit/releases/download/v${LIVEKIT_SERVER_VERSION}/livekit_${LIVEKIT_SERVER_VERSION}_linux_$(dpkg --print-architecture).tar.gz" \
        | tar -xz -C /usr/local/bin livekit-server \
    && rm -rf /var/lib/apt/lists/*
 
# Python deps first for layer caching.
COPY pyproject.toml uv.lock /agent/
WORKDIR /agent
RUN uv sync --frozen --no-dev
 
# Agent code + runtime assets.
COPY app /agent/app
COPY agent_desc.txt /agent/agent_desc.txt
COPY db/schema.sql /agent/db/schema.sql
COPY ui /agent/ui
COPY start.sh /agent/start.sh
 
WORKDIR /agent