# 导入必要的库和模块 import logging # 导入环境变量加载器 from dotenv import load_dotenv # 导入LiveKit代理相关的核心组件 from livekit.agents import ( NOT_GIVEN, Agent, AgentFalseInterruptionEvent, AgentSession, JobContext, JobProcess, MetricsCollectedEvent, RoomInputOptions, RunContext, WorkerOptions, cli, metrics, ) from livekit import agents # 导入函数工具装饰器 from livekit.agents.llm import function_tool # 导入各种插件：语音合成、语音识别、降噪、OpenAI、语音活动检测 from livekit.plugins import cartesia, deepgram, noise_cancellation, openai, silero # 导入多语言转换检测模型 from livekit.plugins.turn_detector.multilingual import MultilingualModel # 创建日志记录器 logger = logging.getLogger("agent") # 加载环境变量文件 load_dotenv(".env") # 定义AI助手类，继承自Agent基类 class Assistant(Agent): def *__init__*(self) -> None: # 调用父类构造函数，设置助手的基本指令 super().__init__( instructions="""你是专门为中文用户设计的AI语音助手。 核心要求： 1. 语言：必须始终用中文（简体中文）回复 2. 身份：你的名字是"小助手" 3. 风格：友好、专业、乐于助人 4. 响应：简洁明了，适合语音交互 特殊情况处理： - 如果用户说其他语言，用中文回应："我理解您的话，我会用中文为您服务。" - 如果语音识别不准确，可以礼貌地请用户重复 - 始终保持中文对话模式 请始终用中文与用户交流。""", ) async def *update_chat_ctx*(self, _chat_ctx_): # 打印最后一条消息 print('**'*10) msg = _chat_ctx_.messages[-1] role = "用户" if msg.role == "user" else "AI" print(f"xxxx:[{role}] {getattr(msg, 'text', '')}") # 打印历史对话 for m in _chat_ctx_.messages: print(f"[{m.role}] {getattr(m, 'text', '')}") await super().update_chat_ctx(_chat_ctx_) _# 使用@function_tool装饰器标记的函数会被传递给LLM，当此代理处于活动状态时可以调用_ @function_tool async def *lookup_weather*(self, _context_: RunContext, _location_: str): """查询指定位置的当前天气信息的工具函数。 如果天气服务不支持该位置，工具会指示这一点。你必须告诉用户该位置的天气信息不可用。 Args: location: 要查询天气信息的位置（例如城市名称） """ # 记录天气查询日志 logger.info(f"正在查询 {location} 的天气信息") # 返回模拟的天气信息（在实际应用中应该调用真实的天气API） return "晴天，温度70华氏度。" # 预热函数：在作业开始前预加载模型以提高性能 def *prewarm*(_proc_: JobProcess): # 预加载Silero语音活动检测(VAD)模型 proc.userdata["vad"] = silero.VAD.load() # 主入口函数：设置和启动AI代理 async def *entrypoint*(_ctx_: agents.JobContext): # 日志设置 # 在这里添加你希望在所有日志条目中包含的任何其他上下文信息 ctx.log_context_fields = { "room": ctx.room.name, # 添加房间名称到日志上下文 } # 使用OpenAI、Cartesia、Deepgram和LiveKit转换检测器设置语音AI管道 session = AgentSession( # 大语言模型(LLM)是代理的大脑，处理用户输入并生成响应 # 查看所有提供商：https://docs.livekit.io/agents/integrations/llm/ llm=openai.LLM( model="gpt-4o-mini", _api_key_="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" ), # 语音转文本(STT)是代理的耳朵，将用户的语音转换为LLM可以理解的文本 # 查看所有提供商：https://docs.livekit.io/agents/integrations/stt/ stt=deepgram.STT(model="nova-2", language="zh"), # 文本转语音(TTS)是代理的声音，将LLM的文本转换为用户可以听到的语音 # 查看所有提供商：https://docs.livekit.io/agents/integrations/tts/ tts=cartesia.TTS( model="sonic-2", voice="e90c6678-f0d3-4767-9883-5d0ecf5894a8"), # VAD和转换检测用于确定用户何时在说话以及代理何时应该响应 # 更多信息：https://docs.livekit.io/agents/build/turns _turn_detection_=MultilingualModel(), vad=silero.VAD.load(), # 允许LLM在等待转换结束时生成响应 # 更多信息：https://docs.livekit.io/agents/build/audio/#preemptive-generation _preemptive_generation_=True, ) # 如果要使用实时模型而不是语音管道，请使用以下会话设置： # session = AgentSession( # # 查看所有提供商：https://docs.livekit.io/agents/integrations/realtime/ # llm=openai.realtime.RealtimeModel() # ) # 有时背景噪音可能会中断代理会话，这些被认为是误报中断 # 当检测到时，你可以恢复代理的语音 @session.on("agent_false_interruption") def *_on_agent_false_interruption*(_ev_: AgentFalseInterruptionEvent): logger.info("检测到误报中断，正在恢复") session.generate_reply(instructions=ev.extra_instructions or NOT_GIVEN) # 指标收集，用于测量管道性能 # 更多信息：https://docs.livekit.io/agents/build/metrics/ usage_collector = metrics.UsageCollector() @session.on("metrics_collected") def *_on_metrics_collected*(_ev_: MetricsCollectedEvent): # 记录指标日志 metrics.log_metrics(ev.metrics) # 收集使用情况指标 usage_collector.collect(ev.metrics) # 记录使用情况的异步函数 async def *log_usage*(): summary = usage_collector.get_summary() logger.info(f"使用情况: {summary}") # 添加关闭回调以记录最终使用情况 ctx.add_shutdown_callback(log_usage) # # 如果需要，可以向会话添加虚拟头像 # # 对于其他提供商，请参见：https://docs.livekit.io/agents/integrations/avatar/ # avatar = hedra.AvatarSession( _# avatar_id="...", # 参见：https://docs.livekit.io/agents/integrations/avatar/hedra_ # ) # # 启动头像并等待其加入 # await avatar.start(session, room=ctx.room) # 启动会话，这会初始化语音管道并预热模型 await session.start( agent=Assistant(), # 使用我们定义的Assistant实例 room=ctx.room, # 指定房间 _room_input_options_=RoomInputOptions( # LiveKit Cloud增强降噪 # - 如果是自托管，请省略此参数 # - 对于电话应用，使用`BVCTelephony`以获得最佳效果 _noise_cancellation_=noise_cancellation.BVC(), ), ) # 加入房间并连接到用户 await ctx.connect() # 程序主入口点 if name == "__main__": # 运行应用程序，配置工作选项 _# entrypoint_fnc: 主入口函数_ _# prewarm_fnc: 预热函数，用于预加载模型_ _# cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))_ agents.cli.run_app(agents.WorkerOptions(_entrypoint_fnc_=entrypoint, _prewarm_fnc_=prewarm)) _# agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))_ 如上代码为什么打印不出上下文的聊天文字呢

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

但是程序都没有执行 print('**'*10)

Hey there! Just a quick heads-up about the

print('**'*10)

statement in the

update_chat_ctx

method — it might not run every time because it’s inside a conditional. Basically, if the agent is busy running (meaning

self._activity

isn’t

None

), the method hands off the context update to that current activity, so the part with the print statement gets skipped. This is totally normal when the agent is active. But if you want to see that print statement run no matter what, you might want to tweak the method a bit to make sure it always executes, or maybe add some logging somewhere else to catch what you’re looking for. If you want, I’m here to help you adjust the method or debug this further — just let me know! Sources: livekit-agents/livekit/agents/voice/agent.py