https://livekit.io logo
Join Slack
Powered by
session = AgentSession[HelperSessionInfo]( llm=gro...
# ask-ai
d
session = AgentSession[HelperSessionInfo]( llm=groq.LLM( model="", temperature=0.1 ), how do i use my own llm
t
Thanks for reaching out, Billy. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
I'm glad you're interested in using your own LLM in LiveKit! You have two friendly options to get started: either use existing provider-specific plugins or create a completely custom LLM integration tailored just for you. 1. Using Pre-Built LLM Integrations: If your LLM has a direct integration, like OpenAI, Azure OpenAI, or Ollama, you can easily use the static methods from the 
openai.LLM
class. For example, here’s how you can use Ollama:
Copy code
from livekit.plugins import openai

llm_instance = openai.LLM.with_ollama(
    model="llama3.1",
    base_url="<http://localhost:11434/v1>"
)
2. Implementing a Custom LLM: If your LLM doesn’t have a ready-made integration, no worries! You can build your own by extending the base 
LLM
class and implementing the 
chat()
method. This method will take care of conversation history, tool calls, and API connection options. Here’s a simple example to get you started with a custom LLM:
Copy code
from livekit.agents import llm

class CustomLLM(llm.LLM):
    def __init__(self, model: str, api_key: str, base_url: str = None):
        super().__init__()
        self._model = model
        self._api_key = api_key
        self._base_url = base_url

    @property
    def model(self):
        return self._model

    def chat(self, *, chat_ctx, tools=None, conn_options=None, parallel_tool_calls=True, tool_choice=None, extra_kwargs=None):
        # Implement integration with your custom LLM API
        return CustomLLMStream(self, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options, extra_kwargs=extra_kwargs or {})
If you'd like, I’m here to walk you through more details on building a custom LLM, including how to handle streaming responses and tool calls. Just let me know—I'm happy to help! Sources: Large language model (LLM) integrations | LiveKit Docs | Ollama integration guide | LiveKit Docs | livekit-agents/livekit/agents/llm/llm.py
d
So i cant just plug in the custom and it will work the same...theres more configuration?
t
You're absolutely right! Using a custom LLM does require a bit of extra setup, but I'm here to help guide you through it. You'll want to create a custom integration by extending the base 
LLM
class and defining exactly how your LLM interacts with LiveKit. Here's what that usually looks like: • Implementing the 
chat()
method to manage the conversation flow and API calls. • Making sure your LLM supports the streaming interface so you can get incremental responses smoothly. • Configuring any tool calls and connection options based on your needs. If your LLM offers an OpenAI-compatible API, there might be a simpler route using existing methods like 
with_ollama
. Please feel free to reach out if you’d like me to walk you through setting this up—I’m happy to help!
d
I want to use Moonshot, Groq currently has a model but I get a bunch of API requests errors and im barely using it
t
I'd love to help you with the API request errors you're seeing with the Groq model. Could you please share a bit more detail? For example: • The exact error messages or codes you’re encountering • When these errors happen (like during initialization or while sending requests) • Any specific settings or parameters you’re using Understanding these details will really help us figure out what’s going on and get you the right solution. Thanks for sharing!
d
class GenerateMockup(AgentTask[str]): def __init__(self, chat_ctx=None,): super().__init__( instructions="""Generate Mockup""", chat_ctx=chat_ctx, ) async def on_enter(self) -> None: logger.info("🎯 Generate Mockup task started") await self.session.generate_reply( instructions="Say: 'Please give me a moment while I prepare your mockup image.'" ) ctx = RunContext result = await generate(ctx) if result: await send_sms_with_image(ctx) self.complete("Please check your messages, I sent the mockup") how do i pass the RunContext ctx to the function generate?
t
I'm glad you asked about passing the 
RunContext
to the 
generate
function! You can easily do this by defining your function as a tool using the 
@function_tool
decorator. The great part is that the LiveKit framework will automatically inject the 
RunContext
whenever the function is triggered, making your workflow smoother. Here’s a simple example to guide you:
Copy code
from livekit.agents import RunContext
from livekit.tools import function_tool

@function_tool()
async def generate(context: RunContext):
    # Access session or other context properties
    await context.session.generate_reply(instructions="Generating mockup...")
    # Perform your logic here
    return "Mockup generated successfully"
So, when you call 
generate(ctx)
, the 
RunContext
will be passed in automatically, as long as the function is decorated properly. If anything feels unclear or you want to dive deeper, just let me know—I’m here to help! Sources: Tool definition and use | LiveKit Docs | Tool definition and use | LiveKit Docs
d
Ok but I want to call it on enter, so its automatically called without any user involvement on_enter...(code) await self...(code) await generate(ctx)
t
I’m happy to help you with calling the 
generate
function automatically within the 
on_enter
method! You can simply pass the 
RunContext
instance directly to the function. Since 
on_enter
is part of the task lifecycle, it’s a great place to create or retrieve the 
RunContext
and pass it explicitly. Here’s a friendly example to guide you:
Copy code
class GenerateMockup(AgentTask[str]):
    def __init__(self, chat_ctx=None):
        super().__init__(
            instructions="Generate Mockup",
            chat_ctx=chat_ctx,
        )

    async def on_enter(self) -> None:
        <http://logger.info|logger.info>("🎯 Generate Mockup task started")
        await self.session.generate_reply(
            instructions="Say: 'Please give me a moment while I prepare your mockup image.'"
        )

        # Create or retrieve the RunContext
        ctx = RunContext(session=self.session)

        # Call the generate function
        result = await generate(ctx)

        if result:
            await send_sms_with_image(ctx)
            self.complete("Please check your messages, I sent the mockup")
With this approach, you’re explicitly creating or retrieving the 
RunContext
and passing it to the 
generate
function. If you have any questions or want me to walk you through any part, just let me know—I’m here to help!
d
async def generate_mock() -> Optional[str]: """Generate a mock-up""" logger.info("👀 Starting mock image generation") imgs = ctx.session.userdata.cloudinary_images if not imgs: logger.warning("👀 No images available for mock-up generation") return None
t
It looks like your 
generate_mock
function is trying to access 
ctx.session.userdata.cloudinary_images
, but the 
ctx
(RunContext) isn’t being passed into the function explicitly. To help get this working smoothly, you’ll want to make sure that 
ctx
is included as a parameter when you call the function. Here’s a friendly example of how you can adjust your function:
Copy code
async def generate_mock(ctx: RunContext) -> Optional[str]:
    """Generate a mock-up"""
    <http://logger.info|logger.info>("👀  Starting mock image generation")

    imgs = ctx.session.userdata.cloudinary_images

    if not imgs:
        logger.warning("👀  No images available for mock-up generation")
        return None

    # Add your logic for generating the mock-up here
    return "Mock-up generated successfully"
And when you call this function, just be sure to pass the 
RunContext
instance like this:
Copy code
ctx = RunContext(session=self.session)
result = await generate_mock(ctx)
If anything’s unclear or you’d like a hand with the next steps, please don’t hesitate to ask—I’m here to help!
d
ctx = RunContext[HelperSessionInfo](session=self.session) return self.generate_mockup(ctx) would this work? passing the data class
t
Absolutely, your idea of using 
RunContext[HelperSessionInfo]
to pass the data class is on the right track! Just make sure that 
HelperSessionInfo
is well-defined and includes the attributes your function needs, like 
cloudinary_images
. Here’s a simple example to help you visualize it:
Copy code
from livekit.agents import RunContext

# Define your data class
@dataclass
class HelperSessionInfo:
    cloudinary_images: list[str] | None = None

# Create the RunContext with the data class
ctx = RunContext[HelperSessionInfo](session=self.session)

# Pass the context to your function
result = await self.generate_mockup(ctx)
Also, it’s important to make sure that the 
userdata
in your session is set up with an instance of 
HelperSessionInfo
when the session starts. For example:
Copy code
userdata = HelperSessionInfo(cloudinary_images=["image1.jpg", "image2.jpg"])
session = AgentSession(userdata=userdata)
If you have any questions or need a hand with anything else, just let me know—I’m here to help!
d
RunContext.__init__() missing 2 required keyword-only arguments: 'speech_handle' and 'function_call'", "level": "ERROR", "name": "livekit.agents", "pid": 56, "job_id": "AJ_YZj36gpCMcPV", "timestamp": "2025-09-18T165338.637265+00:00"}
t
It looks like the error is happening because the 
RunContext
initialization is missing the required keyword arguments 
speech_handle
and 
function_call
. These are essential when you create a 
RunContext
instance. No worries though! To fix this, you just need to include these arguments when you initialize the 
RunContext
. Here’s a simple example to guide you:
Copy code
from livekit.agents import RunContext

# Assuming you have these objects available
speech_handle = None  # Replace with actual speech handle if applicable
function_call = None  # Replace with actual function call if applicable

# Create the RunContext
ctx = RunContext[HelperSessionInfo](
    session=self.session,
    speech_handle=speech_handle,
    function_call=function_call
)

# Pass the context to your function
result = await self.generate_mockup(ctx)
If you don’t have specific values for 
speech_handle
and 
function_call
right now, it’s perfectly okay to set them to 
None
as placeholders. Feel free to reach out if you want me to walk you through this or if anything else comes up—I’m here to help!
d
class HelperAgent(Agent): def __init__(self, tools=None): instructions = f""" FOLLOW THESE RULES STRICTLY: 1. ONLY CALL FUNCTIONS DO NOT READ OUTLOUD ASTERIX OR FUNCTION NAME 2. CURRENT DATE/TIME: {datetime.now()} 3. CALL hangup_call function when appropriate """ super().__init__(instructions=instructions) async def on_enter(self) -> None: collect_result = await CollectPaintingDetails(chat_ctx=self.chat_ctx) if collect_result: ask_result = await AskForPhotos(chat_ctx=self.chat_ctx) if ask_result: logger.info("✈️ Generate Mockup") return HelperAgent how do i get out of the on_enter and return back to the helperagent
t
I’m happy to help you understand how to exit the 
on_enter
method and pass control back to the 
HelperAgent
. You can do this by simply calling 
self.complete()
or making sure the task finishes its lifecycle. Since the 
on_enter
method is part of the task lifecycle, just returning from it won’t automatically switch back to the agent. You’ll want to explicitly manage that flow. Here’s a friendly example to guide you:
Copy code
class HelperAgent(Agent):
    def __init__(self, tools=None):
        instructions = f"""
        FOLLOW THESE RULES STRICTLY:
        1. ONLY CALL FUNCTIONS DO NOT READ OUTLOUD ASTERIX OR FUNCTION NAME
        2. CURRENT DATE/TIME: {datetime.now()}
        3. CALL hangup_call function when appropriate
        """
        super().__init__(instructions=instructions)

    async def on_enter(self) -> None:
        collect_result = await CollectPaintingDetails(chat_ctx=self.chat_ctx)
        if collect_result:
            ask_result = await AskForPhotos(chat_ctx=self.chat_ctx)
            if ask_result:
                <http://logger.info|logger.info>("✈️  Generate Mockup")
                # Complete the task and return control
                self.complete("Task completed, returning to HelperAgent")
By using 
self.complete()
, you’re letting the system know the task is finished, so the agent can move forward smoothly. If anything isn’t clear or you want to dive deeper, just let me know—I’m here to help!
d
and how do i make it call function_tool generate_mockup
t
To help you make the 
on_enter
method call the 
function_tool
decorated 
generate_mockup
function, you can simply invoke it directly, as long as the function is properly defined and registered. Here’s a friendly guide on how you can do it: 1. First, make sure your 
generate_mockup
function is decorated with 
@function_tool
and accepts a 
RunContext
parameter:
Copy code
from livekit.tools import function_tool

@function_tool()
async def generate_mockup(context: RunContext):
    <http://logger.info|logger.info>("Generating mockup...")
    # Your logic here
    return "Mockup generated successfully"
2. Then, inside your 
on_enter
method, create or retrieve the 
RunContext
and call the function like this:
Copy code
class HelperAgent(Agent):
    async def on_enter(self) -> None:
        # Create or retrieve the RunContext
        ctx = RunContext(session=self.session, speech_handle=None, function_call=None)

        # Call the generate_mockup function
        result = await generate_mockup(ctx)

        if result:
            <http://logger.info|logger.info>("Mockup generation completed")
            self.complete("Task completed, returning to HelperAgent")
This way, the 
generate_mockup
function will be called smoothly as part of the 
on_enter
method, ensuring the task lifecycle is handled properly. If you have any questions or need a hand with anything else, please don’t hesitate to ask—I’m here to help!
d
async def on_enter(self) -> None: self.session.generate_reply(instructions="Call generate_mockup function")
t
I understand you're looking to have the 
on_enter
method trigger the 
generate_mockup
function indirectly by guiding the agent to call it. A great way to do this is by using the 
generate_reply
method to provide that instruction. Just keep in mind, for this to work smoothly, the agent needs to be set up to understand and carry out such instructions. Here’s a simple example to illustrate:
Copy code
async def on_enter(self) -> None:
    # Instruct the agent to call the generate_mockup function
    await self.session.generate_reply(instructions="Call generate_mockup function")
To make sure everything works as expected: • Confirm that the 
generate_mockup
function is registered as a tool using the 
@function_tool
decorator. • Verify that your agent is configured to parse and act on instructions like "Call generate_mockup function." If your agent isn’t set up to automatically interpret these instructions yet, no worries—you can always call the function directly inside the 
on_enter
method, just like in the earlier examples. Feel free to reach out if you'd like me to walk you through that or if you have any other questions!
Open in Slack
PreviousNext
Powered by