how can we easily switch our dispatch jobs that send jobs to livekit cloud to the locally running livekit instance for testing

Thanks for reaching out, Bobvi. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

1. (Optional) Use the LiveKit CLI to target local server `````` can i keep my prod env variables like this

You can keep your prod environment variables unchanged while using the LiveKit CLI to target a local server — don’t need to modify prod env vars to test locally. Recommended options (pick one that fits your workflow): • Use a separate CLI project pointing at local a. Add a local project without touching prod:

lk project add local --url <http://localhost:7880> --api-key devkey --api-secret secret
     lk project set-default local

1. Switch back to your prod project when done:

lk project set-default <prod-project-name>

• Override only for your local/test session ◦ Export local env vars in your shell or test script so they don’t affect prod processes:

export LIVEKIT_URL=<http://localhost:7880>
    export LIVEKIT_API_KEY=devkey
    export LIVEKIT_API_SECRET=secret
    # run tests or CLI commands here

• Close the shell or unset the vars after testing:

unset LIVEKIT_URL LIVEKIT_API_KEY LIVEKIT_API_SECRET

• Use per-environment config files or container / process-level envs ◦ Keep prod vars in your production environment, and use a docker-compose file, .env.test, or CI job to inject local test credentials for the test process only. If you tell me how you currently run dispatch jobs (direct env vars, containers, or CI), I’ll give the exact commands or a minimal script to keep prod untouched while you test locally.

are task groups in python stable is that the best way to build them can you show me the reccoemended pattern for python agents that do data capture

You’re asking whether Python task groups are stable, whether they’re the best approach for data-capture agents, and for a recommended pattern. Here’s a concise answer and a recommended pattern. Stability and suitability • Task groups are currently experimental — the API may change. • They are still the recommended pattern for multi-step data capture because they: ◦ enforce ordered steps, ◦ allow users to revisit/correct earlier steps, ◦ share conversation context across tasks, ◦ return typed/structured results that simplify downstream processing. Recommended minimal pattern for a Python data-capture agent • Use TaskGroup to compose small, focused AgentTask classes (one per field/step). • Share a dataclass for the final typed result. • Summarize chat context when helpful (or disable summarization with

summarize_chat_ctx=False

if you need full history). • Keep tasks small, validate inputs inside tasks, and complete each task with a typed result. Example (concise):

from dataclasses import dataclass
from livekit.agents import AgentTask, Agent, function_tool, RunContext
from livekit.agents.beta.workflows import TaskGroup

@dataclass
class UserInfo:
    name: str
    email: str
    phone: str

class GetNameTask(AgentTask[str]):
    def __init__(self):
        super().__init__(instructions="Collect full name.")
    async def on_enter(self):
        await self.session.generate_reply(instructions="Please tell me your full name.")
    @function_tool()
    async def set_name(self, context: RunContext, name: str):
        self.complete(name)

class GetEmailTask(AgentTask[str]):
    def __init__(self):
        super().__init__(instructions="Collect email.")
    async def on_enter(self):
        await self.session.generate_reply(instructions="What is your email?")
    @function_tool()
    async def set_email(self, context: RunContext, email: str):
        # validate email here if desired
        self.complete(email)

class GetPhoneTask(AgentTask[str]):
    def __init__(self):
        super().__init__(instructions="Collect phone.")
    async def on_enter(self):
        await self.session.generate_reply(instructions="Phone number, please.")
    @function_tool()
    async def set_phone(self, context: RunContext, phone: str):
        self.complete(phone)

class DataCaptureAgent(Agent):
    async def on_enter(self):
        tg = TaskGroup(summarize_chat_ctx=True)
        tg.add(lambda: GetNameTask(), id="name")
        tg.add(lambda: GetEmailTask(), id="email")
        tg.add(lambda: GetPhoneTask(), id="phone")

        results = await tg
        user = UserInfo(
            name=results.task_results["name"],
            email=results.task_results["email"],
            phone=results.task_results["phone"],
        )
        await self.session.generate_reply(instructions=f"Thanks, {user.name}. Saved.")

When not to use TaskGroup • If your flow is extremely simple (single prompt/response), TaskGroup may be overkill. • If you require a fully stable API for production-critical code and cannot accept future API changes, consider implementing a small custom step manager instead. If you want, tell me: • whether you need production-grade API stability now, • how many steps and what types of validations you need, and I’ll suggest (a) a TaskGroup-first implementation with migration tips for future API changes, or (b) a simple stable custom-step pattern you can use instead. Sources: Tasks and task groups | LiveKit docs

is there anyway to use the pytest framework ona. task group agent im having trouble wiht that

Thanks — I can help with pytest + TaskGroup. I’m responding to your question about using pytest to test a TaskGroup-based Python agent. Quick answer: yes — pytest works. LiveKit provides testing helpers (AgentSession, LLM mocks) and patterns to simulate multi-turns for TaskGroup flows, but I’ll need a bit more from you to pinpoint the problem. Can you share: • A minimal test snippet that’s failing (your pytest test + agent/task group code), and • The error or unexpected behavior you see (traceback, assertion failures, or what the agent does/doesn’t do)? If you prefer, tell me how you’re invoking the agent in tests (single run vs repeated session.run calls) and whether you’re using a real LLM or a mocked/sandbox LLM — I’ll reply with a focused example or a debugging checklist based on that. Sources: Prompting guide | LiveKit docs | Testing and evaluation | LiveKit docs | Tasks and task groups | LiveKit docs

show me how to pytest an agent that implements task group

Acknowledging your request: you want a clear pytest example that tests a Python agent implementing TaskGroup. Below is a minimal, practical test pattern you can adapt. 1) Minimal agent (data-capture TaskGroup) — put in my_agent.py

from dataclasses import dataclass
from livekit.agents import Agent, AgentTask, function_tool
from livekit.agents.beta.workflows import TaskGroup

@dataclass
class UserInfo:
    name: str
    email: str

class GetNameTask(AgentTask[str]):
    def __init__(self):
        super().__init__(instructions="Collect name.")
    async def on_enter(self):
        await self.session.generate_reply(instructions="What's your name?")
    @function_tool()
    async def set_name(self, name: str):
        self.complete(name)

class GetEmailTask(AgentTask[str]):
    def __init__(self):
        super().__init__(instructions="Collect email.")
    async def on_enter(self):
        await self.session.generate_reply(instructions="What's your email?")
    @function_tool()
    async def set_email(self, email: str):
        self.complete(email)

class DataCaptureAgent(Agent):
    async def on_enter(self):
        tg = TaskGroup(summarize_chat_ctx=True)
        tg.add(lambda: GetNameTask(), id="name")
        tg.add(lambda: GetEmailTask(), id="email")

        results = await tg
        user = UserInfo(
            name=results.task_results["name"],
            email=results.task_results["email"],
        )
        await self.session.generate_reply(instructions=f"Thanks {user.name}.")

2) Pytest test (uses AgentSession + openai plugin) — put in test_agent.py

import pytest
from livekit.agents import AgentSession
from livekit.plugins import openai
from my_agent import DataCaptureAgent

@pytest.mark.asyncio
async def test_task_group_data_capture():
    async with (
        openai.LLM(model="gpt-4o-mini") as llm,
        AgentSession(llm=llm) as session,
    ):
        await session.start(DataCaptureAgent())

        # Turn 1: user greets -> agent should ask for name
        r1 = await session.run(user_input="Hi")
        await r1.expect.next_event().is_message(role="assistant").judge(
            llm, intent="Ask for user's name."
        )

        # Turn 2: supply name -> expect function/tool call to set_name OR task progression
        r2 = await session.run(user_input="My name is Alice")
        # If your task uses a function call to capture the field:
        await r2.expect.next_event().is_function_call(name="set_name", arguments={"name": "Alice"})
        await r2.expect.next_event().is_message(role="assistant").judge(
            llm, intent="Ask for user's email."
        )

        # Turn 3: supply email -> expect set_email call and final thank-you message
        r3 = await session.run(user_input="<mailto:alice@example.com|alice@example.com>")
        await r3.expect.next_event().is_function_call(name="set_email", arguments={"email": "<mailto:alice@example.com|alice@example.com>"})
        await r3.expect.next_event().is_message(role="assistant").judge(
            llm, intent="Thank the user and finish."
        )

        r3.expect.no_more_events()

Notes and troubleshooting tips • Use repeated

session.run(...)

calls to simulate multi-turn interaction; TaskGroup preserves context across turns. • Tests can assert either function calls (preferred for typed task results) or assistant messages with

.judge(...)

. • If you use a mocked/sandbox LLM, ensure it behaves consistently for

.judge()

checks, or assert raw events (function calls/messages) instead. • Enable verbose test logging for debugging: LIVEKIT_EVALS_VERBOSE=1 uv run pytest -s -o log_cli=true test_agent.py If you share your failing test or the error you’re seeing, I’ll adapt this example to match your exact test harness and agent implementation. Sources: Testing and evaluation | LiveKit docs

i have an agent that goes into 4 differnet task groups ask 19 questions total, how would i make a test for this to make sure the llm si doing the right tool calls and conversational pattern

Thanks — I’m responding to your question about testing an agent that runs across 4 TaskGroups and asks 19 questions total. Below is a focused testing strategy and a scalable pytest pattern to assert tool calls and conversational flow. Testing strategy (high level) • Break the end-to-end flow into logical turns that mirror user interactions. Use repeated session.run(...) calls to simulate turns. • Assert function/tool calls for each task step (preferred) and use message intent checks for conversational prompts. • Test each TaskGroup in isolation (unit-style) and one end-to-end integration test that stitches groups together. • Use a table-driven approach (list of expected events) so the test stays readable and maintainable. Pattern: table-driven pytest for many steps 1. Define a list of expected events in order. Each entry describes: ◦ user_input (what test sends) ◦ expected_event: "message" or "function_call" ◦ for messages: expected intent or text check ◦ for function_call: function name and expected arguments 1. Iterate over the list, calling session.run(user_input=...) and asserting the next events. Minimal example skeleton (adapt to your functions/names)

import pytest
from livekit.agents import AgentSession
from livekit.plugins import openai
from my_agent import ComplexAgent  # your agent

# Define expected sequence across all 19 questions
EXPECTED_FLOW = [
    # Turn 1
    {"user_input": "Hi", "expect": {"type": "message", "intent": "Greet and start group1"}},
    {"user_input": "Alice", "expect": {"type": "function", "name": "set_name", "args": {"name": "Alice"}}},
    # Turn 2
    {"user_input": "<mailto:alice@example.com|alice@example.com>", "expect": {"type": "function", "name": "set_email", "args": {"email": "<mailto:alice@example.com|alice@example.com>"}}},
    # ...
    # Continue for all 19 question interactions
]

@pytest.mark.asyncio
async def test_complex_agent_full_flow():
    async with (
        openai.LLM(model="gpt-4o-mini") as llm,
        AgentSession(llm=llm) as session,
    ):
        await session.start(ComplexAgent())

        for step in EXPECTED_FLOW:
            r = await session.run(user_input=step["user_input"])

            e = step["expect"]
            if e["type"] == "function":
                # assert next event is the function call with args
                await r.expect.next_event().is_function_call(name=e["name"], arguments=e["args"])
                # optionally assert the following assistant message (prompt for next)
                # await r.expect.next_event().is_message(role="assistant").judge(llm, intent=e.get("next_intent"))
            else:
                # message assertion — either judge intent or check text contains
                await r.expect.next_event().is_message(role="assistant").judge(llm, intent=e["intent"])

How to scale and keep tests maintainable • Group tests by TaskGroup: one test per TaskGroup that exercises the questions inside that group. This narrows failure scope when something breaks. • Keep EXPECTED_FLOW definitions in fixtures or separate YAML/JSON files to ease updates. • For repetitive validations, write helpers: ◦ assert_function_call(result, name, args) ◦ assert_message_intent(result, llm, intent) • Use mocked or deterministic LLMs for stable judge checks, or prefer function-call assertions (less brittle). Isolation vs integration • Unit tests: Start the agent and directly invoke TaskGroup tasks (if your framework exposes task entry points) or run only the portion that launches a TaskGroup and simulate inputs for that group. • Integration test: Run the full flow once to ensure groups chain correctly and context carries across. Debugging tips • Run pytest with verbose/logging: LIVEKIT_EVALS_VERBOSE=1 uv run pytest -s -o log_cli=true • If .judge() is flaky, assert raw function calls and exact assistant messages instead. • If ordering issues occur, verify you’re consuming expected events in the same call where they are produced — TaskGroup may produce a function_call then a follow-up message in the same turn. If you want, share: • the names of your task function calls and one example of the agent’s expected message for a question, and I’ll draft a concrete EXPECTED_FLOW array and a complete test file tailored to your agent.

after 20-30 jobs my livekit agent goes silent on node js the stt fully breaks what happens there

Thanks — I’m responding to your issue where the Node.js agent’s STT breaks and the agent goes silent after ~20–30 jobs. Quick diagnosis checklist (most common causes) • STT connection/concurrency limits (LiveKit Inference STT defaults to low active-connection quotas). • STT streams not being closed (zombie streams / WebSocket leaks). • Agent not cleaning up per-job resources (audio buffers, timers, VAD/STT instances). • Single worker overload / memory leak causing degraded processing. • Missing shutdown/drain calls so jobs leave listeners alive. Immediate steps to triage (do these first) 1. Check limits and usage ◦ Verify STT concurrency in your LiveKit Project settings (dashboard). If you hit the 5-connection default, new STT sessions will fail. 1. Inspect logs around the failure window ◦ Enable DEBUG logs for your agent and STT library, reproduce until failure, and capture agent logs and STT init/close traces. 1. Confirm streams are closed per job ◦ Ensure you call the STT provider’s stream end/close method (e.g.,

endInput()

or equivalent) and remove any listeners when a job finishes. 1. Add shutdown/cleanup hooks ◦ Use

ctx.addShutdownCallback(...)

or call

ctx.shutdown()

at job end to close VAD/STT, timers, and sockets. 1. Monitor resource usage while running jobs ◦ Watch memory, CPU, and open file/socket counts across jobs. If memory climbs, you likely have a leak. Code-level checks to add • Always end STT streams:

if (sttStream) {
  try { await sttStream.endInput(); } catch(e) { console.error(e); }
  sttStream = undefined;
}

• Remove event listeners and cancel background tasks:

if (vad) { vad.removeAllListeners(); vad = undefined; }
if (monitorTask) { monitorTask.cancel(); }

• Use shutdown callback to guarantee cleanup:

ctx.addShutdownCallback(async () => {
  // close streams, clear timers, remove listeners
});

Testing and verification • Create a repro script that runs 40 short jobs back-to-back while you: ◦ Log STT open/close events, ◦ Log active sockets/file descriptors, ◦ Capture memory snapshots every N jobs. • If failures correlate to open connections or sockets not closing, fix the cleanup path. Mitigations if you hit service limits • Reuse STT connections where protocol permits instead of opening a new stream per job. • Scale workers horizontally so each worker handles fewer jobs concurrently. • Upgrade plan if you need higher STT concurrency from LiveKit Inference. If you share two things I can give targeted fixes: • a short agent log excerpt around where STT fails (STT init/close and error traces), and • the parts of your code that create/use/close STT streams (or the STT plugin you use), I’ll point to the exact spots to add cleanup, retries, or monitoring. Sources: Silero VAD plugin | LiveKit docs | Quotas and limits | LiveKit docs | Introducing LiveKit Inference: A unified model interface for voice AI | Speech-to-text (STT) models | LiveKit Docs | External data and RAG | LiveKit docs