Feat: Skills for Livekit Agents

examples/voice_agents/skills_agent/agent.py

@@ -0,0 +1,48 @@
+"""Example voice agent that uses the skills system.
+
+This agent demonstrates two ways to use skills:
+1. Pre-activated skills (weather) — available immediately
+2. Registry-based skills (calendar) — the LLM can activate on demand
+
+Run with:
+    python agent.py console
+"""
+
+from pathlib import Path
+
+from livekit.agents import Agent, AgentSession, JobContext, JobProcess, cli
+from livekit.agents.skills import SkillRegistry, load_skill_from_directory
+
+SKILLS_DIR = Path(__file__).parent / "skills"
+
+
+def prewarm(proc: JobProcess) -> None:
+    pass
+
+
+async def entrypoint(ctx: JobContext) -> None:
+    await ctx.connect()
+
+    # Load the weather skill directly (pre-activated)
+    weather_skill = load_skill_from_directory(SKILLS_DIR / "weather")
+
+    # Create a registry with the calendar skill (LLM can activate on demand)
+    registry = SkillRegistry()
+    calendar_skill = load_skill_from_directory(SKILLS_DIR / "calendar")
+    registry.register(calendar_skill)
+
+    agent = Agent(
+        instructions=(
+            "You are a helpful voice assistant. You start with weather capabilities. "
+            "If the user asks about their calendar or scheduling, activate the calendar skill."
+        ),
+        skills=[weather_skill],
+        skill_registry=registry,
+    )
+
+    session = AgentSession()
+    await session.start(agent=agent, room=ctx.room)
+
+
+if __name__ == "__main__":
+    cli.run_app(prewarm=prewarm, entrypoint=entrypoint)

examples/voice_agents/skills_agent/skills/calendar/skill.md

@@ -0,0 +1,8 @@
+---
+name: calendar
+description: Manage calendar events, check schedules, and book meetings
+---
+
+You can help users manage their calendar. Use the available calendar tools
+to check schedules and create new events. Always confirm the date and time
+with the user before booking.

examples/voice_agents/skills_agent/skills/calendar/tools.py

@@ -0,0 +1,25 @@
+from livekit.agents.llm import function_tool
+
+
+@function_tool
+async def check_schedule(date: str) -> str:
+    """Check the user's schedule for a given date.
+
+    Args:
+        date: The date to check (e.g. "2024-03-15" or "tomorrow").
+    """
+    # In a real implementation, this would query a calendar API
+    return f"You have 2 meetings on {date}: standup at 9am and design review at 2pm."
+
+
+@function_tool
+async def create_event(title: str, date: str, time: str) -> str:
+    """Create a new calendar event.
+
+    Args:
+        title: The title of the event.
+        date: The date for the event.
+        time: The time for the event.
+    """
+    # In a real implementation, this would create an event via calendar API
+    return f"Created event '{title}' on {date} at {time}."

examples/voice_agents/skills_agent/skills/weather/skill.md

@@ -0,0 +1,8 @@
+---
+name: weather
+description: Get current weather for a city
+---
+
+You can help users check the weather. Use the get_weather tool to look up
+current conditions for any city. Provide temperature, conditions, and a
+brief summary.

examples/voice_agents/skills_agent/skills/weather/tools.py

@@ -0,0 +1,12 @@
+from livekit.agents.llm import function_tool
+
+
+@function_tool
+async def get_weather(city: str) -> str:
+    """Get the current weather for a city.
+
+    Args:
+        city: The name of the city to check weather for.
+    """
+    # In a real implementation, this would call a weather API
+    return f"The weather in {city} is 72°F and sunny with light winds."

livekit-agents/livekit/agents/__init__.py

@@ -20,7 +20,7 @@
 
 import typing
 
-from . import cli, inference, ipc, llm, metrics, stt, tokenize, tts, utils, vad, voice
+from . import cli, inference, ipc, llm, metrics, skills, stt, tokenize, tts, utils, vad, voice
 from ._exceptions import (
     APIConnectionError,
     APIError,
@@ -58,6 +58,7 @@
     function_tool,
 )
 from .plugin import Plugin
+from .skills import Skill, SkillRegistry
 from .types import (
     DEFAULT_API_CONNECT_OPTIONS,
     NOT_GIVEN,
@@ -193,6 +194,9 @@ def __getattr__(name: str) -> typing.Any:
     "PlayHandle",
     "FlushSentinel",
     "LanguageCode",
+    "Skill",
+    "SkillRegistry",
+    "skills",
     "io",
     "avatar",
     "cli",

livekit-agents/livekit/agents/skills/__init__.py

@@ -0,0 +1,9 @@
+from .loader import load_skill_from_directory
+from .registry import SkillRegistry
+from .skill import Skill
+
+__all__ = [
+    "Skill",
+    "SkillRegistry",
+    "load_skill_from_directory",
+]

livekit-agents/livekit/agents/skills/loader.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import importlib.util
+import re
+import sys
+from pathlib import Path
+
+from ..llm.tool_context import Tool, find_function_tools
+from ..log import logger
+from .skill import Skill
+
+_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
+_FIELD_RE = re.compile(r"^(\w+):\s*(.+)$", re.MULTILINE)
+
+
+def _parse_skill_md(text: str) -> tuple[dict[str, str], str]:
+    """Parse a skill.md file into (frontmatter fields, body instructions).
+
+    Raises:
+        ValueError: If the file has no valid frontmatter block.
+    """
+    match = _FRONTMATTER_RE.match(text)
+    if not match:
+        raise ValueError("skill.md must start with a --- frontmatter block")
+
+    fields = dict(_FIELD_RE.findall(match.group(1)))
+    body = text[match.end() :].strip()
+    return fields, body
+
+
+def load_skill_from_directory(path: str | Path) -> Skill:
+    """Load a skill from a directory containing ``skill.md`` and optionally ``tools.py``.
+
+    The ``skill.md`` uses markdown with frontmatter::
+
+        ---
+        name: calendar
+        description: Manage calendar events and scheduling
+        ---
+
+        You can help users manage their calendar. Use the available
+        calendar tools to book, check, and cancel meetings.
+
+    Args:
+        path: Path to the skill directory.
+
+    Returns:
+        A fully constructed :class:`Skill` instance.
+
+    Raises:
+        FileNotFoundError: If the directory or ``skill.md`` does not exist.
+        ValueError: If required fields are missing.
+    """
+    skill_dir = Path(path)
+    if not skill_dir.is_dir():
+        raise FileNotFoundError(f"Skill directory not found: {skill_dir}")
+
+    md_path = skill_dir / "skill.md"
+    if not md_path.exists():
+        raise FileNotFoundError(f"skill.md not found in {skill_dir}")
+
+    text = md_path.read_text()
+    fields, instructions = _parse_skill_md(text)
+
+    name = fields.get("name")
+    if not name:
+        raise ValueError(f"skill.md in {skill_dir} must define 'name' in frontmatter")
+
+    description = fields.get("description", "")
+
+    if not instructions:
+        raise ValueError(f"skill.md in {skill_dir} must have instructions after the frontmatter")
+
+    # load tools from tools.py if present
+    tools: list[Tool] = []
+    tools_path = skill_dir / "tools.py"
+    if tools_path.exists():
+        tools = list(_load_tools_from_file(tools_path))
+
+    return Skill(name=name, description=description, instructions=instructions, tools=tools)
+
+
+def _load_tools_from_file(path: Path) -> list[Tool]:
+    """Dynamically import a Python file and extract function tools from it."""
+    module_name = f"_livekit_skill_{path.parent.name}_{id(path)}_tools"
+
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    if spec is None or spec.loader is None:
+        logger.warning(f"Could not load tools from {path}")
+        return []
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+
+    try:
+        spec.loader.exec_module(module)
+    except Exception as e:
+        logger.exception("Error loading tools from %s", path)
+        sys.modules.pop(module_name, None)
+        raise ImportError(f"Error loading tools from {path}") from e
+
+    return list(find_function_tools(module))

livekit-agents/livekit/agents/skills/meta_tools.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from ..llm.tool_context import FunctionTool, function_tool
+
+if TYPE_CHECKING:
+    from ..voice.agent import Agent
+
+
+def create_skill_meta_tools(agent: Agent) -> list[FunctionTool]:
+    """Create meta-tools for LLM self-activation of skills.
+
+    Returns ``activate_skill`` and ``deactivate_skill`` function tools
+    that are bound to the given agent.
+    """
+
+    @function_tool
+    async def activate_skill(skill_name: str) -> str:
+        """Activate a skill from the registry, adding its tools and instructions.
+
+        Args:
+            skill_name: Name of the skill to activate.
+        """
+        from .registry import SkillRegistry
+
+        registry: SkillRegistry | None = getattr(agent, "_skill_registry", None)
+        if registry is None:
+            return "Error: No skill registry configured."
+
+        skill = registry.get(skill_name)
+        if skill is None:
+            available = ", ".join(registry.available_skills.keys())
+            return f"Error: Skill '{skill_name}' not found. Available: {available}"
+
+        active_names = [s.name for s in getattr(agent, "_active_skills", [])]
+        if skill_name in active_names:
+            return f"Skill '{skill_name}' is already active."
+
+        await agent.add_skill(skill)
+
+        tool_names = [t.id for t in skill.tools]
+        tool_list = ", ".join(tool_names) if tool_names else "(none)"
+        return (
+            f"Activated skill '{skill_name}'.\n"
+            f"New tools available: {tool_list}\n"
+            f"Instructions: {skill.instructions}"
+        )
+
+    @function_tool
+    async def deactivate_skill(skill_name: str) -> str:
+        """Deactivate an active skill, removing its tools and instructions.
+
+        Args:
+            skill_name: Name of the skill to deactivate.
+        """
+        active_names = [s.name for s in getattr(agent, "_active_skills", [])]
+        if skill_name not in active_names:
+            active = ", ".join(active_names) if active_names else "(none)"
+            return f"Error: Skill '{skill_name}' is not active. Active skills: {active}"
+
+        await agent.remove_skill(skill_name)
+        return f"Deactivated skill '{skill_name}'."
+
+    return [activate_skill, deactivate_skill]