从零开始构建实时AI语音助手：LiveKit Agents 完整指南

LiveKit Agents 是一个功能强大的开源框架，专为构建实时语音和视频AI代理而设计。它能够帮助开发者快速创建能够通过音频和视频与用户进行实时互动的智能代理应用。无论你是想构建智能客服机器人、语音助手、虚拟主播还是其他需要实时交互的AI应用，LiveKit Agents 都能为你提供完整的技术解决方案。这个项目基于 LiveKit 的 WebRTC 技术栈，拥有超过一万颗星标，充分证明了其在开发者社区中的认可度和活跃程度。本文将带你从环境搭建开始，逐步深入了解 LiveKit Agents 的核心功能，并通过实际代码示例帮助你掌握这个框架的使用方法。

环境搭建与项目初始化

在开始使用 LiveKit Agents 之前，我们需要先完成开发环境的准备工作。整个环境搭建过程可以分为以下几个步骤：安装 Python 环境、创建虚拟环境、安装依赖包以及配置必要的API密钥。让我们逐一进行详细讲解。

首先，确保你的系统中已经安装了 Python 3.9 或更高版本。你可以通过以下命令检查当前的 Python 版本：

python --version
# 如果显示的版本低于 3.9，建议升级 Python

接下来，创建一个新的项目目录并进入该目录，然后使用 pip 创建虚拟环境。虚拟环境能够帮助你隔离项目依赖，避免不同项目之间的包冲突问题：

# 创建项目目录
mkdir livekit-agents-demo
cd livekit-agents-demo

# 创建虚拟环境（推荐使用 venv）
python -m venv venv

# 激活虚拟环境
# 在 Linux 或 macOS 上：
source venv/bin/activate
# 在 Windows 上：
# venv\Scripts\activate

激活虚拟环境后，你需要安装 LiveKit Agents 的核心依赖包。LiveKit Agents 提供了模块化的安装方式，你可以根据需要选择安装不同的组件。对于基础的语音代理功能，以下核心包是必需的：

# 安装 LiveKit Agents 核心包
pip install livekit-agents

# 安装必要的语音处理插件
pip install livekit-plugins-openai
pip install livekit-plugins-silero

# silero 是一个优秀的语音活动检测（VAD）插件

在安装完依赖后，你还需要配置一些环境变量。LiveKit Agents 需要连接到 LiveKit 服务器才能正常工作，因此你需要准备以下信息：LiveKit 服务器的 URL 地址、API Key 和 API Secret。你可以选择使用 LiveKit Cloud（托管服务）或自行部署 LiveKit 服务器。对于初学者来说，使用 LiveKit Cloud 会更加简单快捷。

创建一个名为 .env 的文件来存储这些配置信息：

# .env 文件内容
LIVEKIT_URL=wss://your-livekit-server.livekit.cloud
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret

# 如果你使用 OpenAI 的 GPT 模型，还需要配置：
OPENAI_API_KEY=your_openai_api_key

在实际开发中，建议使用 python-dotenv 包来加载这些环境变量：

# 安装 dotenv 包
pip install python-dotenv

# 在代码中加载环境变量
from dotenv import load_dotenv
load_dotenv()

核心概念与架构解析

在深入代码编写之前，理解 LiveKit Agents 的核心概念和架构设计是非常重要的。这将帮助你在实际开发中更加得心应手，也能让你更好地利用框架提供的各项功能。

LiveKit Agents 的架构基于几个关键组件构建而成。第一个核心概念是 Agent（代理）。在 LiveKit Agents 的设计哲学中，Agent 是能够响应用户输入并执行特定任务的实体。每个 Agent 都可以配置特定的指令（instructions），定义其行为模式和响应逻辑。Agent 可以处理多种类型的输入，包括文本、音频和视频，使其成为构建多模态应用的理想选择。

第二个核心概念是 Worker（工作者）和 Job（任务）。LiveKit Agents 使用 Worker-Job 模式来管理代理的部署和扩展。当有新的用户连接或任务触发时，系统会自动启动相应的 Worker 来处理这些请求。这种设计使得系统能够根据负载动态扩展，非常适合生产环境中的高并发场景。

第三个核心概念是 Plugin（插件）系统。LiveKit Agents 采用了插件化的架构设计，将不同的功能模块（如语音识别、语音合成、大语言模型等）封装为可插拔的组件。这种设计带来了极大的灵活性，开发者可以根据项目需求选择不同的插件组合，甚至可以开发自己的插件来扩展框架功能。

第四个核心概念是 Pipeline（管道）。Pipeline 定义了用户输入如何被处理以及如何生成响应的完整流程。一个典型的 Pipeline 包括语音活动检测（VAD）、语音识别（STT）、大语言模型（LLM）处理和语音合成（TTS）等环节。LiveKit Agents 允许你自定义这些环节的实现方式，让你能够根据具体场景选择最合适的技术方案。

第五个核心概念是 Room（房间）。Room 是 LiveKit 中用于管理实时会话的核心单元。在 LiveKit Agents 中，每个 Agent 实例都会在一个 Room 中运行，与连接到该 Room 的用户进行实时互动。Room 提供了音视频传输、参与者管理、事件处理等功能，是构建实时交互应用的基础。

从零开始创建你的第一个 Agent

现在你已经了解了基本概念，接下来让我们开始创建你的第一个 LiveKit Agent。我们将构建一个简单的语音助手，它能够接收用户的语音输入，使用大语言模型进行处理，然后生成语音响应。

首先，创建一个名为 agent.py 的文件，这是我们主要的应用逻辑所在：

# agent.py
from livekit import agents
from livekit.agents import JobContext, JobRequest
from livekit.plugins import openai

# 定义 Agent 类
class MyAssistant(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="你是一个友好的AI助手。请用简洁明了的方式回答用户的问题。"
        )

    async def on_user_enter(self, participant):
        """当用户进入房间时调用"""
        await self.session.generate_reply(
            content="你好！我是你的AI助手，有什么可以帮助你的吗？"
        )

    async def on_user_speak(self, frame):
        """当用户说话时调用"""
        pass

async def request_function(req: JobRequest):
    """任务请求处理函数"""
    ctx = await req.accept(
        identity="assistant",
        name="AI Assistant",
        metadata="agent-instance"
    )

    # 创建并启动 Agent
    assistant = MyAssistant()
    await assistant.run(ctx)

在上面的代码中，我们定义了一个继承自 agents.Agent 的类，并重写了几个关键的生命周期方法。on_user_enter 方法在用户进入房间时被调用，我们在这里让 Agent 主动打招呼。instructions 参数定义了 Agent 的系统提示词，用于指导大语言模型的行为。

接下来，我们需要创建主程序入口来启动这个 Agent：

# main.py
from livekit import agents
from livekit.agents import cli
from agent import request_function

if __name__ == "__main__":
    cli.run_app(
        agents.WorkerOptions(
            request_function=request_function,
        )
    )

运行这个程序需要先确保 LiveKit 服务器正在运行。你可以使用以下命令来启动 Agent：

python main.py

当 Agent 成功启动后，你应该能在 LiveKit 的仪表板中看到新的连接。客户端可以通过 WebRTC 连接到对应的 Room，与 Agent 进行语音对话。

深入理解 Session 和上下文管理

Session 是 LiveKit Agents 中另一个非常重要的概念。它代表了 Agent 与用户之间的一个会话上下文，包含了对话历史、状态信息等内容。正确理解和使用 Session 能够帮助你构建更加智能和个性化的应用。

在之前的例子中，我们使用了 await self.session.generate_reply() 来生成回复。generate_reply 是一个非常实用的方法，它会等待用户说完话后，使用当前的对话历史调用大语言模型生成回复。下面让我们深入了解一下 Session 的更多功能：

class AdvancedAssistant(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="""你是一个专业的技术顾问，专注于软件开发和AI技术。
            请用专业但易懂的语言回答问题。
            如果用户询问代码相关问题，请提供示例代码。"""
        )

    async def on_user_speak(self, frame):
        """处理用户语音输入"""
        pass

    async def process_user_message(self, text: str):
        """处理用户发送的文本消息"""
        # 将用户消息添加到对话历史
        self.session.conversation_history.add_user_message(text)

        # 检查用户意图
        if "帮助" in text:
            await self.session.generate_reply(
                content="我可以帮助你完成以下任务：\n1. 回答技术问题\n2. 解释代码逻辑\n3. 提供编程建议\n请告诉我你需要哪方面的帮助。"
            )
        elif "再见" in text:
            await self.session.generate_reply(
                content="很高兴为你服务！再见，有任何问题随时找我。"
            )
        else:
            # 使用大语言模型生成智能回复
            await self.session.generate_reply()

在上面的代码中，我们演示了如何手动管理对话历史。通过 session.conversation_history，我们可以向对话历史中添加消息，也可以清空历史或进行其他操作。这种细粒度的控制在某些场景下非常有用，比如实现多轮对话的上下文管理。

Session 还提供了许多其他有用的属性和方法。例如，你可以使用 session.user_data 来存储与当前用户相关的自定义数据：

async def on_user_enter(self, participant):
    """初始化用户会话数据"""
    self.session.user_data = {
        "name": participant.name,
        "join_time": datetime.now(),
        "interaction_count": 0
    }
    await self.session.generate_reply(
        content=f"欢迎 {participant.name}！很高兴见到你。"
    )

async def on_user_message(self, text: str):
    """处理用户消息"""
    # 增加交互计数
    self.session.user_data["interaction_count"] += 1

    # 根据交互次数调整回复策略
    if self.session.user_data["interaction_count"] > 10:
        await self.session.generate_reply(
            content="看来我们聊得很开心！你还有什么其他问题吗？"
        )
    else:
        await self.session.generate_reply()

语音处理管道的配置与优化

LiveKit Agents 的强大之处在于其灵活的语音处理管道配置。你可以根据应用场景选择不同的语音识别（STT）、大语言模型（LLM）和语音合成（TTS）插件，从而构建最适合你需求的解决方案。

首先让我们看看如何配置使用 OpenAI 的 Whisper 模型进行语音识别：

from livekit.plugins import openai as openai_plugin
from livekit.agents import stt, tts

# 配置 Whisper STT
whisper_stt = openai_plugin.STT(
    model="whisper-1",
    language="zh"  # 设置为中文
)

# 配置 OpenAI TTS
openai_tts = openai_plugin.TTS(
    model="tts-1",
    voice="alloy"  # 选择语音音色
)

# 在 Agent 中使用这些配置
class ConfiguredAgent(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="你是一个智能助手。",
            stt=whisper_stt,
            tts=openai_tts
        )

除了 OpenAI，LiveKit Agents 还支持多种其他插件。例如，你可以使用 Groq、Anthropic、Google 等提供商的模型服务。下面是一个使用多种插件配置的示例：

# 使用不同的插件组合
from livekit.plugins import groq, anthropic, silero

# 配置 Silero VAD（语音活动检测）
# VAD 用于检测用户何时开始和停止说话
silero_vad = silero.VAD(
    threshold=0.5,
    min_speech_duration=0.1,
    min_silence_duration=0.5
)

# 配置使用 Claude 模型的 Agent
class ClaudeAgent(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="你是一个知识渊博的助手。",
            llm=anthropic.LLM(
                model="claude-3-sonnet-20240229"
            ),
            vad=silero_vad
        )

# 配置使用 Groq 快速推理的 Agent
class FastAgent(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="你需要快速响应用户的问题。",
            llm=groq.LLM(
                model="llama-3.1-70b-versatile"
            )
        )

语音活动检测（VAD）是语音交互中非常关键的组件。Silero 是一个高性能的开源 VAD 模型，它能够准确地检测语音的起止点。正确配置 VAD 参数对于提升用户体验至关重要：

# VAD 参数详解
silero_vad = silero.VAD(
    # threshold: 语音检测的置信度阈值，值越高要求越高
    threshold=0.5,

    # min_speech_duration: 最小语音持续时间（秒）
    # 用于过滤掉短暂的噪音
    min_speech_duration=0.1,

    # min_silence_duration: 最小静音持续时间（秒）
    # 用于判断用户是否已经说完话
    min_silence_duration=0.5,

    # speech_pad: 语音边缘填充时间（秒）
    # 在检测到的语音前后添加额外的缓冲
    speech_pad=0.1
)

如果你需要对 VAD 进行更精细的控制，可以使用 VADOptions 类：

from livekit.agents import VADOptions

custom_vad_options = VADOptions(
    min_speech_duration=0.2,  # 最小语音时长
    min_silence_duration=0.8,  # 静音检测时长
    speech_pad=0.15,  # 语音填充
    max_singleUtterance_duration=30.0,  # 最大单次发言时长
    window_duration=60.0,  # 分析窗口大小
)

构建多轮对话系统

在实际应用中，单轮对话往往无法满足复杂场景的需求。我们需要构建能够进行多轮交互的对话系统，让 Agent 能够理解上下文、记住之前的对话内容，并基于此提供更加智能的响应。

让我创建一个更完善的多轮对话 Agent：

import json
from datetime import datetime

class ConversationalAgent(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="""你是一个专业的旅行顾问助手。
            你的职责是帮助用户规划旅行行程。
            请询问用户的旅行偏好，包括目的地、时间预算等信息。
            当收集到足够信息后，提供完整的旅行建议。"""
        )
        self.conversation_state = {
            "stage": "greeting",  # greeting -> collecting -> planning -> complete
            "destination": None,
            "duration": None,
            "budget": None,
            "travelers": None,
            "preferences": []
        }

    async def on_user_enter(self, participant):
        """用户进入时的欢迎流程"""
        await self.session.generate_reply(
            content=f"你好！我是你的专属旅行顾问 {participant.name}。"
            "很高兴为你服务！请问你想去哪里旅行呢？"
        )
        self.conversation_state["stage"] = "collecting"

    async def on_user_message(self, text: str):
        """处理用户输入的文本消息"""
        state = self.conversation_state
        response = ""

        if state["stage"] == "collecting":
            # 收集目的地信息
            if not state["destination"] and "日本" in text:
                state["destination"] = "日本"
                response = "日本是个很棒的选择！是去东京、大阪还是北海道呢？"
            elif not state["destination"] and "泰国" in text:
                state["destination"] = "泰国"
                response = "泰国很好，去曼谷还是普吉岛？"
            elif not state["destination"]:
                response = "请问你想到哪个国家或城市旅行呢？"

            # 收集行程天数
            elif "天" in text:
                days = self._extract_number(text)
                if days:
                    state["duration"] = days
                    response = f"{days}天的行程安排很充实！"
                    "你的预算是多少呢？（经济型/舒适型/豪华型）"

            # 收集预算信息
            elif "经济" in text:
                state["budget"] = "经济型"
                response = "明白了，你偏好经济实惠的选择。"
                "请问有几个人一起出行呢？"
            elif "舒适" in text or "豪华" in text:
                state["budget"] = "舒适型" if "舒适" in text else "豪华型"
                response = f"好的，{state['budget']}的旅行安排。"
                "请问有几个人一起出行呢？"

            # 收集人数
            elif state["destination"] and "人" in text:
                travelers = self._extract_number(text)
                if travelers:
                    state["travelers"] = travelers
                    response = await self._generate_travel_plan()

        await self.session.generate_reply(content=response)

    def _extract_number(self, text: str) -> int:
        """从文本中提取数字"""
        for char in text:
            if char.isdigit():
                return int(char)
        return None

    async def _generate_travel_plan(self) -> str:
        """根据收集的信息生成旅行计划"""
        state = self.conversation_state
        self.conversation_state["stage"] = "complete"

        # 根据不同目的地生成建议
        if state["destination"] == "日本":
            plan = f"""
根据你的需求，我为你规划了以下 {state['destination'] }行程：

【行程概览】
- 目的地：{state['destination']}
- 时长：{state['duration']}天
- 预算：{state['budget']}
- 人数：{state['travelers']}人

【推荐路线】
第1-2天：东京游览（浅草寺、涩谷十字路口）
第3-4天：京都古都之旅（清水寺、伏见稻荷）
第5-{state['duration']}天：大阪美食之旅

【预算估算】
总费用约 {state['duration'] * 800 * state['travelers']} 元/人

需要我提供更详细的每日行程安排吗？
"""
        elif state["destination"] == "泰国":
            plan = f"""
根据你的需求，我为你规划了以下 {state['destination'] }行程：

【行程概览】
- 目的地：{state['destination']}
- 时长：{state['duration']}天
- 预算：{state['budget']}
- 人数：{state['travelers']}人

【推荐路线】
第1-2天：曼谷城市游览（大皇宫、考山路）
第3-5天：清迈文艺之旅
第6-{state['duration']}天：普吉岛海滩休闲

【特色体验】
- 泰式按摩
- 夜市美食
- 海岛浮潜

需要我详细介绍某个部分吗？
"""
        else:
            plan = "抱歉，我对目的地了解有限。能告诉我你想去的具体城市吗？"

        return plan

在上面的示例中，我们使用 conversation_state 字典来跟踪对话的各个阶段和收集到的信息。这种状态机模式能够帮助我们管理复杂的多轮对话流程，确保 Agent 能够在正确的时机询问正确的问题，并基于收集到的信息做出恰当的响应。

实现函数调用与工具使用

LiveKit Agents 支持 function calling（函数调用）功能，这是构建强大 AI 应用的关键特性。通过函数调用，Agent 可以执行预定义的函数来获取实时信息、控制外部设备或执行各种操作。

让我们来实现一个支持工具调用的智能助手：

from livekit import agents
from livekit.agents import f

# 定义可用的工具函数
@f推测
def get_weather(location: str) -> str:
    """获取指定位置的天气信息

    Args:
        location: 城市名称，如"北京"、"东京"等
    """
    # 模拟天气数据
    weather_data = {
        "北京": {"temp": 22, "condition": "晴天", "humidity": 45},
        "上海": {"temp": 25, "condition": "多云", "humidity": 60},
        "东京": {"temp": 20, "condition": "小雨", "humidity": 75},
        "纽约": {"temp": 18, "condition": "晴天", "humidity": 50},
    }

    if location in weather_data:
        data = weather_data[location]
        return f"{location}今天天气{data['condition']}，气温{data['temp']}度，湿度{data['humidity']}%。"
    return f"抱歉，暂未获取到{location}的天气信息。"

@f推测
def search_news(topic: str) -> str:
    """搜索指定主题的最新新闻

    Args:
        topic: 新闻主题关键词
    """
    # 模拟新闻数据
    news_data = {
        "AI": [
            "OpenAI 发布 GPT-5，性能大幅提升",
            "Google 推出新版 Gemini AI 助手",
            "Meta 开源 Llama 3 新版本"
        ],
        "科技": [
            "苹果发布 iOS 18 最新测试版",
            "特斯拉全自动驾驶获批新地区",
            "SpaceX 完成星舰第九次试飞"
        ],
        "体育": [
            "世界杯预选赛精彩对决",
            "NBA 新赛季即将开幕",
            "网球大满贯赛事精彩回顾"
        ]
    }

    if topic in news_data:
        articles = news_data[topic]
        result = f"关于{topic}的最新新闻：\n"
        for i, article in enumerate(articles, 1):
            result += f"{i}. {article}\n"
        return result
    return f"暂无{topic}相关的最新新闻。"

@f推测
def set_reminder(time: str, content: str) -> str:
    """设置提醒事项

    Args:
        time: 提醒时间，格式为"YYYY-MM-DD HH:MM"
        content: 提醒内容
    """
    # 模拟设置提醒
    return f"已设置提醒：{time} - {content}"

# 使用这些工具的 Agent
class ToolEnabledAgent(agents.Agent):
    def __init__(self):
        super().__init__(
            instructions="""你是一个智能助手，可以帮助用户查询天气、新闻，
            并设置提醒。请主动了解用户需求并使用适当的工具来帮助他们。
            回答要简洁、有条理。""",
            fnc_ctx=f.create_context(
                get_weather,
                search_news,
                set_reminder
            )
        )

在上面的代码中，我们使用 @f推测 装饰器来声明函数工具。这个装饰器会自动生成函数调用所需的 schema 信息，让大语言模型知道何时以及如何调用这些函数。在 Agent 的初始化中，我们通过 f.create_context() 创建函数调用上下文，并将所有可用的工具传入。

当 Agent 接收到用户请求时，它会自动判断是否需要调用工具。例如，当用户说“北京今天天气怎么样”时，Agent 会识别出这是一个天气查询请求，然后调用 get_weather 函数并将结果返回给用户。

构建 Web 客户端界面

为了与 LiveKit Agents 进行交互，我们需要创建一个客户端应用。LiveKit 提供了 JavaScript SDK，可以让你轻松构建 Web 端的语音交互界面。下面是一个完整的 Web 客户端示例：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>LiveKit 语音助手</title>
    <script src="https://cdn.jsdelivr.net/npm/livekit-client@latest/dist/livekit-client.umd.min.js"></script>
    <style>
        * {
            margin: 0;
            padding: 0;
            box-sizing: border-box;
        }

        body {
            font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
            min-height: 100vh;
            display: flex;
            justify-content: center;
            align-items: center;
            padding: 20px;
        }

        .container {
            background: white;
            border-radius: 20px;
            box-shadow: 0 20px 60px rgba(0, 0, 0, 0.3);
            width: 100%;
            max-width: 500px;
            padding: 40px;
        }

        h1 {
            text-align: center;
            color: #333;
            margin-bottom: 30px;
            font-size: 24px;
        }

        .status-indicator {
            display: flex;
            align-items: center;
            justify-content: center;
            margin-bottom: 30px;
            gap: 10px;
        }

        .status-dot {
            width: 12px;
            height: 12px;
            border-radius: 50%;
            background: #ccc;
            transition: background 0.3s;
        }

        .status-dot.connected {
            background: #4caf50;
            box-shadow: 0 0 10px #4caf50;
        }

        .status-dot.speaking {
            background: #f44336;
            box-shadow: 0 0 15px #f44336;
            animation: pulse 0.5s infinite alternate;
        }

        @keyframes pulse {
            from { transform: scale(1); }
            to { transform: scale(1.2); }
        }

        .status-text {
            color: #666;
            font-size: 14px;
        }

        .controls {
            display: flex;
            flex-direction: column;
            gap: 15px;
        }

        button {
            padding: 15px 30px;
            font-size: 16px;
            border: none;
            border-radius: 10px;
            cursor: pointer;
            transition: all 0.3s;
            font-weight: 600;
        }

        .btn-connect {
            background: #667eea;
            color: white;
        }

        .btn-connect:hover {
            background: #5568d3;
            transform: translateY(-2px);
        }

        .btn-connect:disabled {
            background: #ccc;
            cursor: not-allowed;
            transform: none;
        }

        .btn-mic {
            background: #4caf50;
            color: white;
            display: none;
        }

        .btn-mic.active {
            display: block;
        }

        .btn-mic:hover {
            background: #45a049;
        }

        .btn-mic.speaking {
            background: #f44336;
        }

        .btn-mic.speaking:hover {
            background: #d32f2f;
        }

        .btn-disconnect {
            background: #f44336;
            color: white;
            display: none;
        }

        .btn-disconnect.active {
            display: block;
        }

        .btn-disconnect:hover {
            background: #d32f2f;
        }

        .log-container {
            margin-top: 30px;
            background: #f5f5f5;
            border-radius: 10px;
            padding: 15px;
            max-height: 200px;
            overflow-y: auto;
        }

        .log-title {
            font-size: 12px;
            color: #666;
            margin-bottom: 10px;
            text-transform: uppercase;
        }

        .log-entry {
            font-size: 12px;
            padding: 5px 0;
            border-bottom: 1px solid #e0e0e0;
            color: #333;
        }

        .log-entry:last-child {
            border-bottom: none;
        }

        .log-entry .time {
            color: #999;
            margin-right: 10px;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>🎙️ LiveKit 语音助手</h1>

        <div class="status-indicator">
            <div class="status-dot" id="statusDot"></div>
            <span class="status-text" id="statusText">未连接</span>
        </div>

        <div class="controls">
            <button class="btn-connect" id="btnConnect">连接服务器</button>
            <button class="btn-mic" id="btnMic">按住说话</button>
            <button class="btn-disconnect" id="btnDisconnect">断开连接</button>
        </div>

        <div class="log-container">
            <div class="log-title">会话日志</div>
            <div id="logContent"></div>
        </div>
    </div>

    <script>
        // LiveKit 连接配置
        const LIVEKIT_URL = 'wss://your-livekit-server.livekit.cloud';
        const ROOM_NAME = 'agent-room';
        const TOKEN = 'your-access-token';  // 需要从后端获取

        let room = null;
        let isSpeaking = false;

        // DOM 元素
        const btnConnect = document.getElementById('btnConnect');
        const btnMic = document.getElementById('btnMic');
        const btnDisconnect = document.getElementById('btnDisconnect');
        const statusDot = document.getElementById('statusDot');
        const statusText = document.getElementById('statusText');
        const logContent = document.getElementById('logContent');

        // 日志记录函数
        function addLog(message) {
            const time = new Date().toLocaleTimeString();
            const entry = document.createElement('div');
            entry.className = 'log-entry';
            entry.innerHTML = `<span class="time">${time}</span>${message}`;
            logContent.appendChild(entry);
            logContent.scrollTop = logContent.scrollHeight;
        }

        // 连接 LiveKit 服务器
        async function connectToRoom() {
            try {
                addLog('正在连接到服务器...');

                room = new livekit.Room({
                    adaptiveStream: true,
                    dynacast: true,
                });

                // 设置音频处理
                await room.setParticipantPermissions({
                    canPublish: true,
                    canSubscribe: true,
                    canPublishData: true,
                });

                // 连接到房间
                await room.connect(LIVEKIT_URL, TOKEN);
                addLog('已成功连接到房间');

                // 更新 UI 状态
                statusDot.classList.add('connected');
                statusText.textContent = '已连接';
                btnConnect.style.display = 'none';
                btnMic.classList.add('active');
                btnDisconnect.classList.add('active');

                // 启用麦克风
                await room.localParticipant.setMicrophoneEnabled(true);
                addLog('麦克风已启用');

                // 监听远程参与者（Agent）的音频
                room.on(livekit.RoomEvent.ParticipantConnected, (participant) => {
                    addLog(`${participant.name} 已加入`);
                });

                room.on(livekit.RoomEvent.ParticipantDisconnected, (participant) => {
                    addLog(`${participant.name} 已离开`);
                });

                // 监听音频轨道
                room.on(livekit.RoomEvent.TrackSubscribed, (track, publication, participant) => {
                    if (track.kind === livekit.TrackKind.AUDIO) {
                        const audioElement = track.attach();
                        document.body.appendChild(audioElement);
                        addLog(`正在播放来自 ${participant.name} 的音频`);
                    }
                });

            } catch (error) {
                addLog(`连接失败: ${error.message}`);
                console.error(error);
            }
        }

        // 断开连接
        function disconnectFromRoom() {
            if (room) {
                room.disconnect();
                room = null;
                addLog('已断开连接');

                statusDot.classList.remove('connected', 'speaking');
                statusText.textContent = '未连接';
                btnConnect.style.display = 'block';
                btnMic.classList.remove('active');
                btnDisconnect.classList.remove('active');
            }
        }

        // 事件绑定
        btnConnect.addEventListener('click', connectToRoom);
        btnDisconnect.addEventListener('click', disconnectFromRoom);

        // 麦克风按住说话功能
        btnMic.addEventListener('mousedown', () => {
            isSpeaking = true;
            btnMic.classList.add('speaking');
            statusDot.classList.add('speaking');
            statusText.textContent = '正在说话...';
            addLog('开始录音');
        });

        btnMic.addEventListener('mouseup', () => {
            isSpeaking = false;
            btnMic.classList.remove('speaking');
            statusDot.classList.remove('speaking');
            statusText.textContent = '已连接 - 等待响应';
            addLog('录音结束');
        });

        // 触摸设备支持
        btnMic.addEventListener('touchstart', (e) => {
            e.preventDefault();
            isSpeaking = true;
            btnMic.classList.add('speaking');
            statusDot.classList.add('speaking');
            statusText.textContent = '正在说话...';
            addLog('开始录音');
        });

        btnMic.addEventListener('touchend', (e) => {
            e.preventDefault();
            isSpeaking = false;
            btnMic.classList.remove('speaking');
            statusDot.classList.remove('speaking');
            statusText.textContent = '已连接 - 等待响应';
            addLog('录音结束');
        });
    </script>
</body>
</html>

这个 Web 客户端提供了完整的用户界面，包括连接状态指示、麦克风控制按钮和会话日志。在实际部署时，你需要注意以下几点：首先，需要在服务端生成正确的访问令牌（Token）；其次，Web 应用需要通过 HTTPS 提供访问（因为浏览器对麦克风权限有安全要求）；最后，根据实际需求修改 LiveKit 服务器地址和房间名称。

服务端 Token 生成

为了安全地连接到 LiveKit 房间，我们需要生成访问令牌。这个令牌包含了用户的身份信息和权限设置。让我为你展示如何在 Python 中生成令牌：

“`python

token_generator.py

from livekit import api
from datetime import datetime, timedelta

def generate_agent_token():
“””生成 Agent 使用的访问令牌”””

# Token 配置
api_key = "your_api_key"
api_secret = "your_api_secret"

# Token 有效期设置
now = datetime.now()
expire_at = now + timedelta(hours=24)

# 创建 GrantParticipant token
# 这个 token 用于以参与者的身份加入房间
token = api.AccessToken(api_key, api_secret) \
    .with_identity("ai-assistant") \
    .with_name("AI Assistant") \
    .with_valid_for(timedelta(hours=24)) \
    .with_grants(api.VideoGrants(
        room_join=True,
        room="agent-room",
        can_publish=True,
        can_subscribe=True,
        can_publish_data=True,
    ))

# 生成 JWT token
jwt_token = token.to_jwt()
print(f"生成的 Token: {jwt_token}")

return jwt_token

def generate_user_token(user_identity: str, user_name: str = None):
“””生成用户访问令牌

Args:
    user_identity: 用户唯一标识
    user_name: 用户显示名称
"""

api_key = "your_api_key"
api_secret = "your_api_secret"

token = api

Project: https://github.com/livekit/agents

Stars: 10504