llm_node

lookup(query) -> Optional[str]

store(query, response)

llm_node

async def llm_node(self, chat_ctx: llm.ChatContext, tools: list, model_settings) -> AsyncIterable[llm.ChatChunk]:
    # 1) Build a cache key from the incoming context (message text, metadata, etc.)
    cache_key = self.semantic_cache.build_key(chat_ctx)

    # 2) Try cache hit
    cached = await self.semantic_cache.lookup(cache_key)
    if cached:
        # Return cached text as a single ChatChunk (or stream if you prefer)
        yield llm.ChatChunk(text=cached)
        return

    # 3) No hit — call default LLM behavior and stream results, while capturing them
    collected = []
    async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
        collected.append(chunk.text)
        yield chunk

    # 4) Store combined response in cache
    full_response = "".join(collected)
    await self.semantic_cache.store(cache_key, full_response)

llm_node

lookup(key)

store(key, value)

chat_ctx.messages[-1].content

import asyncio
from typing import AsyncIterable
from livekit import llm  # adjust import to your SDK
from livekit.agents import Agent  # adjust import path
from livekit.agents import llm as llm_types  # for ChatChunk type if needed

# Simple in-memory semantic cache (replace with vector DB or Redis in prod)
class SemanticCache:
    def __init__(self):
        self._store: dict[str, str] = {}

    async def build_key(self, chat_ctx) -> str:
        # Example: use last user message as key (customize with embeddings/metadata)
        return chat_ctx.messages[-1].content.strip()

    async def lookup(self, key: str) -> str | None:
        # Real semantic cache would do similarity search; this is exact match
        await asyncio.sleep(0)  # simulate async I/O
        return self._store.get(key)

    async def store(self, key: str, value: str):
        await asyncio.sleep(0)  # simulate async I/O
        self._store[key] = value

# Example Agent subclass with llm_node override
class CachingAgent(Agent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.semantic_cache = SemanticCache()

    async def llm_node(
        self,
        chat_ctx: llm.ChatContext,
        tools: list,
        model_settings
    ) -> AsyncIterable[llm.ChatChunk]:
        # 1) Build cache key
        cache_key = await self.semantic_cache.build_key(chat_ctx)

        # 2) Try cache hit
        cached = await self.semantic_cache.lookup(cache_key)
        if cached:
            # Return cached response as a single ChatChunk
            yield llm.ChatChunk(text=cached)
            return

        # 3) No cache hit — call default LLM and stream results while collecting
        collected_chunks: list[str] = []
        async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
            # chunk may be str or ChatChunk depending on SDK; handle both
            text = chunk.text if hasattr(chunk, "text") else str(chunk)
            collected_chunks.append(text)
            yield chunk

        # 4) Store the combined response in cache
        full_response = "".join(collected_chunks)
        await self.semantic_cache.store(cache_key, full_response)

conversation_item_added

update_chat_ctx()

conversation_item_added

import asyncio
from livekit.agents import AgentSession, JobContext, llm, inference
from livekit.agents.events import ChatContextEvent

# --- Simple planner tools (replace with real implementations) ---
class WebSearchTool:
    async def search(self, query: str) -> str:
        await asyncio.sleep(0)  # simulate async I/O
        return f"Search results for: {query}"

class SummarizerTool:
    async def summarize(self, text: str) -> str:
        await asyncio.sleep(0)
        return f"Summary: {text[:200]}"

# --- Background planner ---
class BackgroundPlanner:
    def __init__(self, session: AgentSession, planner_llm: inference.LLM):
        self.session = session
        self.planner_llm = planner_llm
        self.search_tool = WebSearchTool()
        self.summary_tool = SummarizerTool()
        # subscribe to chat events
        self.session.event_queue.subscribe(ChatContextEvent)
        self._task = asyncio.create_task(self._loop())

    async def _loop(self):
        async for event in self.session.event_queue.disposable_iter():
            if not isinstance(event, ChatContextEvent) or event.added_item is None:
                continue
            item = event.added_item
            # only react to user messages
            if getattr(item, "role", "") == "user":
                text = item.content
                # run research/summarization in background
                asyncio.create_task(self._research_and_inject(text))

    async def _research_and_inject(self, user_text: str):
        # 1) Research
        results = await self.search_tool.search(user_text)

        # 2) Summarize results using planner LLM (or summarizer tool)
        # Example uses planner_llm.chat() — adapt to your SDK's API
        chat_ctx = llm.ChatContext()
        chat_ctx.add_message(role="system", content="You are a research assistant.")
        chat_ctx.add_message(role="user", content=f"Please summarize these findings:\n\n{results}")

        completion = await self.planner_llm.chat(chat_ctx=chat_ctx)
        planner_summary = completion.message.content if completion and completion.message else await self.summary_tool.summarize(results)

        # 3) Inject the summary into the main agent's chat context
        new_ctx = self.session.chat_ctx.copy()
        new_ctx.add_message(role="system", content=f"[PLANNER UPDATE]: {planner_summary}")
        await self.session.update_chat_ctx(new_ctx)

# --- Example main session setup ---
async def start_session(ctx: JobContext):
    session = AgentSession(
        stt=inference.STT(model="some-stt"),
        llm=inference.LLM(model="gpt-4o-mini"),  # main agent LLM (fast)
        tts=inference.TTS(model="some-tts"),
    )

    # planner uses a separate, stronger model
    planner_llm = inference.LLM(model="gpt-4.1")
    BackgroundPlanner(session=session, planner_llm=planner_llm)

    await session.start(agent=YourMainAgent(), room=ctx.room)

import asyncio
from typing import Any
from livekit.agents import Agent, AgentSession, function_tool
from livekit.agents.llm.async_toolset import AsyncToolset, AsyncRunContext
from livekit.agents import inference, llm

# --- Async toolset ---
class ResearchToolset(AsyncToolset):
    def __init__(self):
        super().__init__(id="research")

    @function_tool()
    async def perform_research(self, ctx: AsyncRunContext, query: str) -> str:
        # progress update visible in chat context while running
        await ctx.update(f"Starting research for: {query}")

        # simulate long-running research (replace with real web search / API calls)
        await asyncio.sleep(2)
        results = f"Fake search results for: {query}"

        await ctx.update("Research complete, summarizing...")
        # simulate summarization
        await asyncio.sleep(1)
        summary = f"Summary of '{query}': {results[:200]}"

        # final return becomes the tool result
        return summary

# --- Main agent (no changes needed for tool calling) ---
class MyAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are a helpful assistant.")
        # Agent-level tools are cancelled on handoff; prefer session-level tools if persistence needed

# --- Start session with async tool registered on the session ---
async def start_session():
    session = AgentSession(
        stt=inference.STT(model="some-stt"),
        llm=inference.LLM(model="gpt-4o-mini"),  # main agent LLM
        tts=inference.TTS(model="some-tts"),
        tools=[ResearchToolset()],  # register async tools on session (persist across handoffs)
    )

    # Start the agent session (replace with your actual session.start usage)
    await session.start(agent=MyAgent(), room="room-name")

# --- How it works ---
# - The LLM (main agent) will only call `perform_research` when it decides the tool is needed
#   (based on prompt/context or explicit user command).
# - The tool runs asynchronously; use `await ctx.update(...)` to send progress updates into chat.
# - You can rely on the LLM to invoke the tool via ReAct-style reasoning, or prompt it explicitly:
#     User: "Research: best restaurants in Seattle"
#   The agent will decide to call the async tool and return the tool result when ready.

# Run example (in real code this is driven by your server framework)
# asyncio.run(start_session())

AsyncToolset

AgentSession

ctx.update()

ResearchToolset

import asyncio
from livekit.agents import Agent, AgentSession, function_tool, llm, inference
from livekit.agents.llm.async_toolset import AsyncToolset, AsyncRunContext
from livekit.agents.events import ChatContextEvent

# --- Async toolset used by the sub-agent ---
class ResearchToolset(AsyncToolset):
    def __init__(self):
        super().__init__(id="research")

    @function_tool()
    async def perform_research(self, ctx: AsyncRunContext, query: str) -> str:
        await ctx.update(f"Planner: starting research for '{query}'")
        await asyncio.sleep(2)  # replace with real web search
        results = f"Fake search results for: {query}"
        await ctx.update("Planner: summarizing results...")
        await asyncio.sleep(1)
        return f"Summary of '{query}': {results[:200]}"

# --- Main agent (fast, responsive) ---
class MainAgent(Agent):
    def __init__(self):
        super().__init__(instructions="You are the primary assistant.")

# --- Sub-agent / Background planner that uses its own LLM and tools ---
class BackgroundPlanner:
    def __init__(self, session: AgentSession, planner_llm: inference.LLM):
        self.session = session
        self.planner_llm = planner_llm
        # Register tools on the session so they persist across handoffs
        # (Session already created with tools=[ResearchToolset()], see start_session)
        self.session.event_queue.subscribe(ChatContextEvent)
        self._task = asyncio.create_task(self._loop())

    async def _loop(self):
        async for ev in self.session.event_queue.disposable_iter():
            if not isinstance(ev, ChatContextEvent) or ev.added_item is None:
                continue
            item = ev.added_item
            # Trigger policy: sub-agent acts only on explicit "PLAN:" commands
            if getattr(item, "role", "") == "user" and item.content.startswith("PLAN:"):
                query = item.content[len("PLAN:"):].strip()
                asyncio.create_task(self._run_planner(query))

    async def _run_planner(self, query: str):
        # Build a small chat context for the planner
        chat_ctx = llm.ChatContext()
        chat_ctx.add_message(role="system", content="You are a planning assistant.")
        chat_ctx.add_message(role="user", content=f"Research: {query}")

        # Call planner's LLM; allow it to call async tools (ResearchToolset)
        # The planner_llm should be configured to allow tool use (handled by session tools)
        completion = await self.planner_llm.chat(chat_ctx=chat_ctx)

        # If planner's LLM directly returns result, inject it; otherwise the async tool's
        # progress/return will already be added to the session via ctx.update() from the tool.
        planner_out = completion.message.content if completion and completion.message else None
        if planner_out:
            new_ctx = self.session.chat_ctx.copy()
            new_ctx.add_message(role="system", content=f"[PLANNER RESULT]: {planner_out}")
            await self.session.update_chat_ctx(new_ctx)

# --- Start session wiring: register async tool on session and start both agents ---
async def start_session():
    session = AgentSession(
        stt=inference.STT(model="some-stt"),
        llm=inference.LLM(model="gpt-4o-mini"),        # main agent LLM (fast)
        tts=inference.TTS(model="some-tts"),
        tools=[ResearchToolset()],                     # session-level async tools
    )

    # Planner uses a separate, more capable LLM
    planner_llm = inference.LLM(model="gpt-4.1")       # planner/sub-agent LLM

    # Attach the background planner (sub-agent)
    BackgroundPlanner(session=session, planner_llm=planner_llm)

    # Start main agent
    await session.start(agent=MainAgent(), room="room-name")

# Example trigger:
# - User sends: "PLAN: best electric cars 2026"
# - Planner sees "PLAN:" command, runs research tool (async), and injects summary when ready.

planner_llm

ctx.update()

ctx.update()