#!/usr/bin/env python3 """LiveKit Voice AI Agent - v5 (配置驱动)""" import asyncio import logging import os import io import wave import hashlib import sqlite3 from typing import Optional, AsyncGenerator, List, Dict, Any, Union from enum import Enum from collections import deque import time import json import yaml from pathlib import Path from dotenv import load_dotenv load_dotenv('/Users/leo/.hermes/workspace/livekit-agents/.env.dev') try: from livekit import rtc from livekit.agents import JobContext, AutoSubscribe except ImportError: rtc = None JobContext = None logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ========== 配置加载 ========== class Config: """统一配置管理""" def __init__(self, config_path: str = None): self.data: Dict[str, Any] = {} if config_path and os.path.exists(config_path): with open(config_path) as f: self.data = yaml.safe_load(f) or {} self._merge_env() def _merge_env(self): """环境变量覆盖""" if os.environ.get("LIVEKIT_URL"): self.data.setdefault("livekit", {})["url"] = os.environ["LIVEKIT_URL"] if os.environ.get("OPENAI_API_KEY"): self.data.setdefault("openai", {})["api_key"] = os.environ["OPENAI_API_KEY"] if os.environ.get("OPENAI_BASE_URL"): self.data.setdefault("openai", {})["base_url"] = os.environ["OPENAI_BASE_URL"] def get(self, key: str, default=None): keys = key.split(".") val = self.data for k in keys: if isinstance(val, dict): val = val.get(k) else: return default return val if val is not None else default @property def agent(self) -> Dict[str, Any]: return self.data.get("agent", {}) @property def livekit(self) -> Dict[str, Any]: return self.data.get("livekit", {}) @property def stt(self) -> Dict[str, Any]: return self.agent.get("stt", {}) @property def tts(self) -> Dict[str, Any]: return self.agent.get("tts", {}) @property def llm(self) -> Dict[str, Any]: return self.agent.get("llm", {}) @property def pipeline(self) -> Dict[str, Any]: return self.agent.get("pipeline", {}) @property def memory(self) -> Dict[str, Any]: return self.agent.get("memory", {}) @property def vad(self) -> Dict[str, Any]: return self.agent.get("vad", {}) @property def barge_in(self) -> Dict[str, Any]: return self.agent.get("barge_in", {}) @property def vision(self) -> Dict[str, Any]: return self.agent.get("vision", {}) # 全局配置 config = Config('/Users/leo/.hermes/workspace/livekit-agents/config.yaml') # ========== 状态枚举 ========== class AgentState(str, Enum): IDLE = "idle" LISTENING = "listening" PROCESSING = "processing" SPEAKING = "speaking" INTERRUPTED = "interrupted" # ========== 第一板：分句合成器 ========== class SentenceSplitter: """分句合成器""" SENTENCE_ENDINGS = set(config.pipeline.get("sentence_boundaries", "。！？；\n")) @classmethod def split_sentences(cls, text: str) -> List[str]: sentences = [] current = "" for char in text: current += char if char in cls.SENTENCE_ENDINGS and current.strip(): sentences.append(current.strip()) current = "" if current.strip(): sentences.append(current.strip()) return sentences @classmethod async def process_stream(cls, llm_stream: AsyncGenerator[str, None]) -> AsyncGenerator[tuple, None]: buffer = "" async for chunk in llm_stream: buffer += chunk if buffer and buffer[-1] in cls.SENTENCE_ENDINGS: yield buffer.strip(), True buffer = "" if buffer.strip(): yield buffer.strip(), True # ========== VAD: TEN VAD ========== class TenVAD: """TEN VAD 包装""" def __init__(self): self.threshold: float = float(config.vad.get("threshold", 0.3)) self.min_speech_ms: float = float(config.vad.get("min_speech_ms", 300)) self.min_silence_ms: float = float(config.vad.get("min_silence_ms", 500)) self._vad = None self._is_speaking = False self._speech_start_time = 0.0 self._silence_start_time = 0.0 self._buffer = bytearray() async def initialize(self): try: import ten_vad self._vad = ten_vad.TenVad(hop_size=256, threshold=self.threshold) logger.info(f"✅ TEN VAD 就绪 (threshold={self.threshold})") except ImportError: logger.warning("⚠️ TEN VAD 未安装") self._vad = None def process_frame(self, audio_data: bytes) -> Optional[bool]: if self._vad is None: return self._energy_detect(audio_data) try: import numpy as np audio_np = np.frombuffer(audio_data, dtype=np.int16)[:256] score, is_speech = self._vad.process(audio_np) now = time.time() * 1000 if is_speech: if not self._is_speaking: self._is_speaking = True self._speech_start_time = now self._buffer = bytearray() return True else: self._buffer.extend(audio_data) self._silence_start_time = 0 else: if self._is_speaking: if now - self._speech_start_time < self.min_speech_ms: self._is_speaking = False self._buffer = bytearray() return False if self._silence_start_time == 0: self._silence_start_time = now if now - self._silence_start_time >= self.min_silence_ms: self._is_speaking = False return False return None except Exception: return None def _energy_detect(self, audio_data: bytes) -> Optional[bool]: try: import numpy as np arr = np.frombuffer(audio_data, dtype=np.int16) return float(np.mean(arr.astype(float) ** 2)) > 100 except: return None def get_buffer(self) -> bytes: return bytes(self._buffer) def reset(self): self._buffer = bytearray() self._is_speaking = False self._silence_start_time = 0 # ========== STT ========== class FunASRSTT: """FunASR STT""" def __init__(self, endpoint: str): self.endpoint = endpoint self._client = None async def initialize(self): import httpx self._client = httpx.AsyncClient(base_url=self.endpoint, timeout=10.0) logger.info(f"🎤 STT 就绪: {self.endpoint}") await self._warmup() async def _warmup(self): try: buf = io.BytesIO() with wave.open(buf, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(b"\x00\x00" * 160) self._client.post("/stream", files={"file": ("test.wav", buf.getvalue(), "audio/wav")}) logger.info("✅ STT 预热完成") except Exception as e: logger.warning(f"STT 预热失败: {e}") async def recognize(self, audio_bytes: bytes) -> str: if len(audio_bytes) < 3200: return "" try: buf = io.BytesIO() with wave.open(buf, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(audio_bytes[:160000]) response = await self._client.post("/stream", files={"file": ("audio.wav", buf.getvalue(), "audio/wav")}) response.raise_for_status() text = response.json().get("text", "") logger.info(f"🎤 识别: {text[:50]}...") return text except Exception as e: logger.error(f"STT 失败: {e}") return "" async def close(self): if self._client: await self._client.aclose() # ========== LLM ========== class TokenCounter: """Token 估算器（中文按字符估算）""" @staticmethod def estimate_tokens(text: str) -> int: """估算文本的 token 数（粗略：1 token ≈ 2 中文字符）""" if not text: return 0 # 中文字符和英文混合估算 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') other_chars = len(text) - chinese_chars # 中文：1字符≈1token，英文：1词≈1.3token return int(chinese_chars + other_chars / 1.3) @staticmethod def count_messages(messages: List[dict]) -> int: """计算消息列表的总 token 数""" total = 0 for msg in messages: content = msg.get("content", "") total += TokenCounter.estimate_tokens(content) # 加上 role 字段的开销 total += len(msg.get("role", "")) // 2 return total class MemoryManager: """Token-aware 记忆管理器""" def __init__(self, config: dict): self.token_limit = config.get("token_limit", 4000) self.chat_history_ratio = config.get("chat_history_ratio", 0.8) self._history: List[dict] = [] self._system_prompt: str = config.get("system_prompt", "你是智能助手。") self._session_id: Optional[str] = None self._persist_path: Optional[str] = None def set_session(self, session_id: str, persist: bool = False): """设置会话 ID，可选持久化""" self._session_id = session_id if persist: self._persist_path = f"/tmp/voice_agent_{session_id}.json" self._load_persisted() def _load_persisted(self): """加载持久化历史""" if self._persist_path and os.path.exists(self._persist_path): try: with open(self._persist_path) as f: data = json.load(f) self._history = data.get("history", []) logger.info(f"📂 加载历史: {len(self._history)} 条") except Exception as e: logger.warning(f"加载历史失败: {e}") def _save_persisted(self): """保存历史到文件""" if self._persist_path: try: with open(self._persist_path, 'w') as f: json.dump({"history": self._history}, f, ensure_ascii=False) except Exception as e: logger.warning(f"保存历史失败: {e}") def add_user_message(self, message: str): """添加用户消息""" self._history.append({"role": "user", "content": message}) self._cleanup() self._save_persisted() def add_assistant_message(self, message: str): """添加工具响应""" self._history.append({"role": "assistant", "content": message}) self._cleanup() self._save_persisted() def get_messages(self) -> List[dict]: """获取用于 LLM 的消息列表""" return [ {"role": "system", "content": self._system_prompt}, *self._history ] def _cleanup(self): """清理历史，确保不超过 token 限制""" # 预留 20% 给系统提示 available = int(self.token_limit * self.chat_history_ratio) # 从旧到新删除，直到满足限制 while len(self._history) > 1: current_tokens = TokenCounter.count_messages(self._history) if current_tokens <= available: break self._history.pop(0) # 删除最早的对话 logger.debug(f"💾 记忆管理: {len(self._history)} 轮, ~{TokenCounter.count_messages(self._history)} tokens") def get_stats(self) -> dict: """获取记忆统计""" return { "session_id": self._session_id, "total_turns": len(self._history) // 2, "tokens_used": TokenCounter.count_messages(self._history), "token_limit": self.token_limit, "persist_path": self._persist_path } def clear(self): """清空记忆""" self._history = [] if self._persist_path and os.path.exists(self._persist_path): os.remove(self._persist_path) logger.info("🗑️ 记忆已清空") class LLMAgent: """LLM Agent - Token-aware 记忆管理""" def __init__(self): self.backend = config.llm.get("backend", "openai") self.model = config.llm.get("model", "agnes-2.5-flash") self._client = None self.memory = MemoryManager(config.agent.get("memory", {})) self._system_prompt = config.agent.get("system_prompt", "你是 Agnes，智能语音助手。") async def initialize(self): from openai import AsyncOpenAI base_url: str = config.get("openai.base_url", "https://api.agnes-ai.cn/v1") api_key: str = config.get("openai.api_key", "") self._client = AsyncOpenAI(api_key=api_key, base_url=base_url) logger.info(f"🤖 LLM 就绪: {self.backend} / {self.model}") async def chat(self, message: str, session_id: Optional[str] = None) -> AsyncGenerator[str, None]: """流式对话，自动管理记忆""" if session_id: self.memory.set_session(session_id) self.memory.add_user_message(message) try: messages = self.memory.get_messages() response = await self._client.chat.completions.create( model=self.model, messages=messages, max_tokens=300, stream=True, ) assistant_response = [] async for chunk in response: if chunk.choices and chunk.choices[0].delta.content: content = chunk.choices[0].delta.content assistant_response.append(content) yield content full_response = "".join(assistant_response) self.memory.add_assistant_message(full_response) except Exception as e: logger.error(f"LLM 失败: {e}") yield "抱歉，我遇到了一点问题。" async def close(self): pass # ========== TTS ========== class EdgeTTS: """Edge TTS""" def __init__(self, voice: str = "zh-CN-XiaoxiaoNeural"): self.voice = voice self._initialized = False async def initialize(self): import edge_tts self._communicate = edge_tts.Communicate self._initialized = True logger.info(f"🔊 TTS 就绪: {self.voice}") async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]: import edge_tts comm = self._communicate(text, self.voice) async for chunk in comm.stream(): if chunk.get("type") == "audio": yield chunk["data"] # ========== 意图判断器 ========== class IntentClassifier: """判断用户话语是否需要 Agent 介入回复""" # 唤醒词模式（可配置） wakeup_patterns = ["小美", "银美", "agnes", "agent"] # 明确提问关键词 question_keywords = ["你觉得", "你怎么看", "帮我", "请问", "你觉得呢", "你怎么认为"] # 附和/确认词（不需要回复） filler_words = {"嗯", "对", "好的", "明白", "哦", "啊", "呵", "嗯嗯", "对的", "是的", "好", "嗯啊"} @classmethod def classify(cls, text: str) -> str: """ 返回: IGNORE | INTERRUPT | FILLER """ text_lower = text.lower().strip() # 1. 检查是否为附和词 if text_lower in cls.filler_words or text_lower.startswith(("嗯", "啊", "哦")): return "FILLER" # 2. 检查唤醒词 for pattern in cls.wakeup_patterns: if pattern in text_lower: return "INTERRUPT" # 3. 检查明确提问 for keyword in cls.question_keywords: if keyword in text_lower: return "INTERRUPT" # 4. 默认：忽略（被动监听） return "IGNORE" # ========== Barge-in: 语义判停（旧版，保留兼容） ========== class BargeInClassifier: """Barge-in 语义判停""" def __init__(self): self._filler_phrases = {"嗯", "对", "好的", "明白", "哦", "啊", "呵", "嗯嗯", "对的"} async def classify(self, transcript: str) -> str: text = transcript.strip().lower() if text in self._filler_phrases: return "filler" if len(text) <= 2: return "filler" try: from openai import AsyncOpenAI client = AsyncOpenAI( api_key=config.get("openai.api_key", ""), base_url=config.get("openai.base_url", "https://api.agnes-ai.cn/v1") ) response = await client.chat.completions.create( model=config.llm.get("model", "agnes-2.5-flash"), messages=[{ "role": "system", "content": "判断用户插话意图，只返回 filler 或 real" }, { "role": "user", "content": text }], max_tokens=10, temperature=0, ) intent = response.choices[0].message.content.strip().lower() return intent if intent in ["filler", "real"] else "real" except Exception as e: logger.error(f"分类器失败: {e}") return "real" # ========== Echo Suppressor ========== class EchoSuppressor: """回声抑制""" def __init__(self): self._current_speech = "" def set_current_speech(self, text: str): self._current_speech = text def clear(self): self._current_speech = "" def is_echo(self, asr_text: str) -> bool: if not self._current_speech or not asr_text: return False clean_play = self._current_speech.replace("。", "").replace("！", "").replace("？", "").strip() clean_asr = asr_text.replace("。", "").replace("！", "").replace("？", "").strip() if len(clean_asr) > 10 and clean_asr in clean_play: return True return False # ========== AudioPlayQueue ========== class AudioPlayQueue: """播放队列""" def __init__(self, sample_rate: int = 24000): self.sample_rate = sample_rate self._queue = deque() self._is_playing = False self._lock = asyncio.Lock() self._cancelled = False self._current_chunk = bytearray() async def enqueue(self, audio_chunk: bytes): async with self._lock: self._queue.append(audio_chunk) if not self._is_playing: self._is_playing = True self._current_task = asyncio.create_task(self._play_loop()) async def clear(self): async with self._lock: self._queue.clear() self._current_chunk.clear() self._cancelled = True async def resume(self): async with self._lock: self._cancelled = False async def _play_loop(self): while self._queue and not self._cancelled: chunk = self._queue.popleft() self._current_chunk.extend(chunk) target_samples = int(self.sample_rate * 0.2) while len(self._current_chunk) >= target_samples: sample_chunk = bytes(self._current_chunk[:target_samples]) self._current_chunk = self._current_chunk[target_samples:] await asyncio.sleep(len(sample_chunk) / (self.sample_rate * 2)) self._is_playing = False self._cancelled = False # ========== 主 Agent (v5 - 配置驱动) ========== class VoiceAgent: """Voice AI Agent v5 - 配置驱动""" def __init__(self): # 组件 self.splitter = SentenceSplitter() self.vad = TenVAD() self.stt = FunASRSTT(config.stt.get("endpoint", "http://localhost:8080")) self.llm = LLMAgent() self.tts = EdgeTTS(config.tts.get("voice", "zh-CN-XiaoxiaoNeural")) if config.tts.get("enabled", True) else None self.classifier = IntentClassifier() self.echo_suppressor = EchoSuppressor() self.play_queue = AudioPlayQueue(sample_rate=config.pipeline.get("sample_rate", 24000)) # 状态 self._state = AgentState.IDLE self._initialized = False self._stop_event = asyncio.Event() self._tts_task = None self._audio_source = None # 话轮检测（新增） self._silence_start_time = None # 静音开始时间 self._current_speaker: Optional[str] = None # 对话记录器（说话人+面部识别+持久化） from dialogue_logger import DialogueLogger self.dialogue_logger = DialogueLogger() self._min_silence_ms = config.vad.get("min_silence_ms", 500) # 持续静音阈值 # 人物档案管理器 from profile_manager import ProfileManager self.profile_manager = ProfileManager() async def initialize(self): await self.vad.initialize() await self.stt.initialize() await self.llm.initialize() if self.tts: await self.tts.initialize() self._initialized = True logger.info("✅ Agent v5 初始化完成 (配置驱动)") @property def state(self) -> AgentState: return self._state async def set_state(self, new_state: AgentState): old_state = self._state self._state = new_state logger.info(f"📊 状态: {old_state.value} → {new_state.value}") async def process_audio_frame(self, audio_data: bytes) -> None: if not self._initialized: return vad_result = self.vad.process_frame(audio_data) current_time = time.time() * 1000 # 毫秒 if vad_result is True: # 有声音：重置静音计时器 self._silence_start_time = None if self._state == AgentState.SPEAKING: asyncio.create_task(self._handle_barge_in()) elif self._state == AgentState.IDLE: await self.set_state(AgentState.LISTENING) elif vad_result is False: # 静音：记录开始时间，持续超过阈值才触发 if self._state == AgentState.LISTENING: if self._silence_start_time is None: self._silence_start_time = current_time # 开始计时 # 检查是否持续静音超过阈值 silence_duration = current_time - self._silence_start_time if silence_duration >= self._min_silence_ms: audio_buffer = self.vad.get_buffer() if len(audio_buffer) > 3200: # 最短语音阈值 await self._handle_speech(audio_buffer) self._silence_start_time = None # 重置 async def _handle_barge_in(self): current_audio = self.vad.get_buffer() filler_text = await self.stt.recognize(current_audio[:8000]) if config.barge_in.get("echo_suppression", True) and self.echo_suppressor.is_echo(filler_text): logger.info("🔇 回声抑制") return if config.barge_in.get("semantic_classifier", True): intent = await self.classifier.classify(filler_text) if intent == "filler": logger.info(f"🔇 附和: '{filler_text}'") return logger.info(f"⏹️ 真打断: '{filler_text}'") await self._cancel_routine() async def _cancel_routine(self): await self.set_state(AgentState.INTERRUPTED) if self._tts_task and not self._tts_task.done(): self._tts_task.cancel() try: await self._tts_task except asyncio.CancelledError: pass await self.play_queue.clear() self.vad.reset() await asyncio.sleep(0.1) self._stop_event.set() await self.set_state(AgentState.IDLE) async def _handle_speech(self, audio_buffer: bytes): await self.set_state(AgentState.PROCESSING) text = await self.stt.recognize(audio_buffer) if not text.strip(): await self.set_state(AgentState.IDLE) self.vad.reset() return # 记录对话（包含说话人信息） current_speaker = self._current_speaker or "unknown" if hasattr(self, 'dialogue_logger'): scene_hash = None if hasattr(self, 'visual_monitor') and self.visual_monitor: scene_hash = self.visual_monitor._last_frame_hash self.dialogue_logger.record_speech( speaker_id=current_speaker, participant_identity=current_speaker, text=text, scene_hash=scene_hash ) # 自动提取事实并更新人物档案 if hasattr(self, 'profile_manager') and config.get("profile_manager.auto_extract_facts", False): asyncio.create_task(self._extract_facts(current_speaker, text)) # 意图判断：是否需要 Agent 介入 intent = self.classifier.classify(text) logger.info(f"🎯 意图判断: '{text}' → {intent}") if intent == "IGNORE": # 不需要回复，仅记录到记忆 await self.llm.memory.add_user_message(text) await self.set_state(AgentState.IDLE) self.vad.reset() return if intent == "FILLER": # 附和词，忽略 logger.info(f"🔇 附和词: '{text}'") await self.set_state(AgentState.IDLE) self.vad.reset() return # INTERRUPT: 需要 Agent 回复 tts_tasks = [] async for sentence, is_complete in self.splitter.process_stream(self.llm.chat(text)): if sentence.strip() and self.tts: await self.set_state(AgentState.SPEAKING) task = asyncio.create_task(self._play_sentence(sentence)) tts_tasks.append(task) if tts_tasks: await asyncio.gather(*tts_tasks, return_exceptions=True) await self.set_state(AgentState.IDLE) async def _play_sentence(self, sentence: str): if not self.tts or not self._audio_source: return try: self.echo_suppressor.set_current_speech(sentence) chunk_count = 0 async for audio_chunk in self.tts.synthesize_stream(sentence): if self._stop_event.is_set() or self._state == AgentState.INTERRUPTED: break self.play_queue._current_chunk.extend(audio_chunk) target_samples = int(self.play_queue.sample_rate * 0.2) while len(self.play_queue._current_chunk) >= target_samples: sample_chunk = bytes(self.play_queue._current_chunk[:target_samples]) self.play_queue._current_chunk = self.play_queue._current_chunk[target_samples:] await self._play_audio(sample_chunk) chunk_count += 1 if self.play_queue._current_chunk: await self._play_audio(bytes(self.play_queue._current_chunk)) self.echo_suppressor.clear() self.play_queue._current_chunk.clear() logger.info(f"✅ 句子播放完成 ({chunk_count} 分片)") except Exception as e: logger.error(f"TTS 播放失败: {e}") self.echo_suppressor.clear() async def _play_audio(self, audio_data: bytes): if not self._audio_source: return try: frame = rtc.AudioFrame(audio_data, self.play_queue.sample_rate, 1, len(audio_data) // 2) await self._audio_source.capture_frame(frame) except Exception as e: logger.debug(f"播放帧错误: {e}") async def interrupt(self): await self._cancel_routine() async def close(self): if self._tts_task and not self._tts_task.done(): self._tts_task.cancel() await self.stt.close() if hasattr(self, 'dialogue_logger'): self.dialogue_logger.close() if hasattr(self, 'profile_manager'): self.profile_manager.close() async def _extract_facts(self, speaker_id: str, text: str) -> None: """异步提取事实并存储到人物档案""" try: from profile_manager import extract_fact_prompt import re prompt = extract_fact_prompt(text) # 调用 LLM（非流式） if self.llm._client: response = await self.llm._client.chat.completions.create( model=self.llm.model, messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=200 ) llm_response = response.choices[0].message.content if llm_response: # 解析 JSON json_match = re.search(r'\[.*\]', llm_response, re.DOTALL) if json_match: facts = json.loads(json_match.group()) for fact in facts: if isinstance(fact, dict) and 'predicate' in fact and 'object' in fact: self.profile_manager.add_fact( subject_id=speaker_id, predicate=fact['predicate'], object=fact['object'], confidence=fact.get('confidence', 0.8) ) logger.info(f"📝 自动提取事实: {speaker_id} -> {fact['predicate']} -> {fact['object']}") except Exception as e: logger.warning(f"⚠️ 事实提取失败: {e}") async def get_recent_context(self, max_chars: int = 500) -> str: """获取最近的对话上下文""" messages = self.llm.memory.get_messages() # 跳过 system prompt context_msgs = messages[1:] if messages and messages[0].get("role") == "system" else messages context = "" for msg in reversed(context_msgs): content = msg.get("content", "") if len(context) + len(content) > max_chars: break context = content + "\n" + context return context.strip() async def generate_quick_response(self, prompt: str) -> Optional[str]: """生成快速响应（用于主动介入）""" if not self.llm._client: return None try: messages = [{"role": "system", "content": config.agent.get("system_prompt", "")}] messages.append({"role": "user", "content": prompt}) response = await self.llm._client.chat.completions.create( model=self.llm.model, messages=messages, max_tokens=100, temperature=0.7, ) return response.choices[0].message.content except Exception as e: logger.error(f"快速响应生成失败: {e}") return None # ========== 视觉监控（语义记忆）========== class VisualMonitor: """基于视频流的语义记忆系统 - 使用WeMM Embedding存储场景""" def __init__(self, agent): self.agent = agent self.frame_interval = int(config.vision.get("frame_interval", 30)) self.wemm_url = os.getenv("WEMM_API_URL", config.vision.get("wemm_api_url", "http://your-gpu-server:8765")) self.wemm_dimension = int(os.getenv("WEMM_DIMENSION", "2048")) # 语义记忆 self.memory_enabled = config.vision.get("memory", {}).get("enabled", True) self.max_entries = config.vision.get("memory", {}).get("max_entries", 100) self.auto_learn = config.vision.get("memory", {}).get("auto_learn", True) self.learn_interval = config.vision.get("memory", {}).get("learn_interval", 300) # 关键帧检测参数 self.change_threshold = float(config.vision.get("change_threshold", 0.15)) # 相似度低于此值视为场景变化 self.min_interval_seconds = int(config.vision.get("min_interval_seconds", 30)) # 最少间隔30秒 # 记忆存储 self._visual_memories = [] self._last_learn_time = 0 self._last_frame_embedding = None self._last_frame_hash = None self._frame_counter = 0 self._source_url = None # 当前图片/视频的原始URL async def initialize(self): """初始化视觉监控""" logger.info(f"🔗 WeMM API: {self.wemm_url}") if self.memory_enabled: logger.info(f"✅ 语义记忆已启用 (最大{self.max_entries}条)") async def analyze_frame(self, frame, source_url: Optional[str] = None): """分析视频帧 - 仅存储关键帧（场景变化时）""" self._source_url = source_url # 按间隔采样 self._frame_counter += 1 if self._frame_counter % self.frame_interval != 0: return current_time = time.time() # 检查时间间隔 if current_time - self._last_learn_time < self.min_interval_seconds: return # 检测场景变化 is_key_frame = await self._is_key_frame(frame) if is_key_frame: await self._learn_scene(frame, source_url=self._source_url) self._last_learn_time = current_time async def _is_key_frame(self, frame) -> bool: """判断是否为关键帧（场景发生变化）""" if self._last_frame_embedding is None: return True # 第一帧总是关键帧 try: import numpy as np from PIL import Image from io import BytesIO # 转换帧为numpy数组 if not hasattr(frame, 'data'): return False img_array = np.frombuffer(frame.data, dtype=np.uint8) img_array = img_array.reshape((frame.height, frame.width, 3)) # 编码当前帧 pil_img = Image.fromarray(img_array) buffer = BytesIO() pil_img.save(buffer, format="JPEG", quality=50) img_bytes = buffer.getvalue() # 计算轻量级哈希作为快速检测 current_hash = hashlib.md5(img_bytes).hexdigest()[:8] # 如果哈希相同，跳过 if current_hash == self._last_frame_hash: return False # 编码当前帧获取embedding current_embedding = await self._encode_image(img_bytes) if current_embedding is None: return False # 计算与上一帧的相似度 similarity = np.dot(current_embedding, self._last_frame_embedding) # 相似度低于阈值视为场景变化 is_changed = similarity < self.change_threshold # 更新最后帧信息 self._last_frame_embedding = current_embedding self._last_frame_hash = current_hash logger.info(f"📊 帧相似度: {similarity:.3f}, 关键帧: {is_changed}") return is_changed except Exception as e: logger.error(f"关键帧检测失败: {e}") return False async def _learn_scene(self, frame, source_url: Optional[str] = None): """学习当前场景并存入记忆""" try: import base64 from io import BytesIO from PIL import Image, ExifTags import numpy as np # 转换帧 if not hasattr(frame, 'data'): return img_array = np.frombuffer(frame.data, dtype=np.uint8) img_array = img_array.reshape((frame.height, frame.width, 3)) pil_img = Image.fromarray(img_array) buffer = BytesIO() pil_img.save(buffer, format="JPEG", quality=85) img_bytes = buffer.getvalue() # 提取EXIF元数据（从原始帧或source_url） exif_data = await self._extract_exif(pil_img, source_url) # 编码图像 embedding = await self._encode_image(img_bytes) if embedding: # 更新最后帧信息 self._last_frame_embedding = embedding self._last_frame_hash = hashlib.md5(img_bytes).hexdigest()[:8] # 提取场景描述（通过VLM API） scene_description = await self._get_scene_description(pil_img, img_bytes) # 人脸快照 face_snapshots = self._detect_faces(pil_img) # 存入记忆（包含所有信息） self._visual_memories.append({ "timestamp": time.time(), "embedding": embedding, "frame_hash": self._last_frame_hash, "source_url": source_url, "scene_description": scene_description, "face_snapshots": face_snapshots, "exif": exif_data }) # 限制记忆数量 if len(self._visual_memories) > self.max_entries: self._visual_memories = self._visual_memories[-self.max_entries:] # 异步提取照片相关的事实（综合EXIF+场景描述+人脸，由VLM判断主题） if hasattr(self.agent, 'profile_manager') and exif_data: asyncio.create_task(self._extract_facts_from_photo( scene_description, exif_data, face_snapshots, source_url )) logger.info(f"📚 已存储关键帧记忆 #{len(self._visual_memories)}") # 记录到对话日志器 if hasattr(self.agent, 'dialogue_logger'): self.agent.dialogue_logger.record_face_snapshot( self._last_frame_hash, embedding, scene_description or f"场景 #{len(self._visual_memories)}" ) except Exception as e: logger.error(f"学习场景失败: {e}") async def _extract_exif(self, pil_img, source_url: Optional[str] = None) -> Dict: """提取EXIF元数据""" from PIL import ExifTags exif_info = {} try: # 尝试从图片对象获取EXIF exif_data = pil_img._getexif() if exif_data: for tag_id, value in exif_data.items(): tag = ExifTags.TAGS.get(tag_id, tag_id) # 跳过二进制数据 if isinstance(value, bytes): continue exif_info[str(tag)] = str(value)[:200] # 限制长度 except Exception: pass # 如果EXIF为空且source_url可用，可以从URL提取 if not exif_info and source_url: try: import urllib.request req = urllib.request.Request(source_url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=5) as resp: from io import BytesIO raw_bytes = resp.read() temp_img = Image.open(BytesIO(raw_bytes)) exif_data = temp_img._getexif() if exif_data: for tag_id, value in exif_data.items(): tag = ExifTags.TAGS.get(tag_id, tag_id) if isinstance(value, bytes): continue exif_info[str(tag)] = str(value)[:200] except Exception as e: logger.debug(f"EXIF提取失败: {e}") return exif_info async def _get_scene_description(self, pil_img, img_bytes: bytes) -> Optional[str]: """调用VLM API获取场景描述""" try: import base64 import aiohttp # 将图片转为base64 img_base64 = base64.b64encode(img_bytes).decode('utf-8') # 调用Agnes AI多模态API async with aiohttp.ClientSession() as session: payload = { "model": config.llm.get("model", "agnes-2.5-flash"), "messages": [{ "role": "user", "content": [ {"type": "text", "text": "请简短描述这张图片的内容（10字以内）"}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_base64}"}} ] }], "max_tokens": 50 } async with session.post( f"{config.get('openai.base_url', 'https://api.agnes-ai.cn/v1')}/chat/completions", json=payload, headers={"Authorization": f"Bearer {config.get('openai.api_key', '')}"} ) as resp: if resp.status == 200: result = await resp.json() description = result.get("choices", [{}])[0].get("message", {}).get("content", "") return description.strip() if description else None except Exception as e: logger.debug(f"VLM场景描述失败: {e}") return None def _detect_faces(self, pil_img) -> List[Dict]: """检测人脸（简化版，使用PIL）""" faces = [] try: import numpy as np # 转为灰度图 gray = pil_img.convert('L') img_array = np.array(gray) # 简单的面部区域检测（基于肤色和轮廓启发式） # 实际项目应使用face_recognition或dlib height, width = img_array.shape # 这里简化处理，返回空列表 # 当安装face_recognition后启用： # import face_recognition # face_locations = face_recognition.face_locations(img_array) # for (y, x2, y2, x1) in face_locations: # faces.append({"x": (x1+x2)//2, "y": (y+y2)//2, "confidence": 0.9}) except Exception as e: logger.debug(f"人脸检测失败: {e}") return faces async def _extract_facts_from_photo(self, scene_description: Optional[str], exif: Dict, face_snapshots: List[Dict], source_url: Optional[str]): """用VLM综合分析照片信息，提取完整事件（Who/What/When/Where/Why/How）""" try: import aiohttp if not hasattr(self.agent, 'profile_manager'): return # 构建综合提示词（要求六要素完整信息） info_parts = [] if scene_description: info_parts.append(f"场景描述: {scene_description}") if exif: key_tags = ['DateTime', 'Make', 'Model', 'GPSInfo', 'ExposureTime', 'ISO', 'FNumber'] for tag in key_tags: if tag in exif: info_parts.append(f"{tag}: {exif[tag]}") if face_snapshots: info_parts.append(f"人脸数: {len(face_snapshots)}") if source_url: info_parts.append(f"来源URL: {source_url}") info_text = "; ".join(info_parts) if info_parts else "无额外信息" prompt = f"""请分析以下照片的综合信息，提取完整事件记录。 照片信息： {info_text} 请提取以下六要素信息： 1. WHO（人物）：照片中涉及的人物是谁？可能的人物身份/职业/关系 2. WHAT（事件）：发生了什么事件？活动内容是什么？ 3. WHEN（时间）：拍摄时间（从EXIF或场景推断） 4. WHERE（地点）：具体地点（从GPS/场景推断，保留原始精度） 5. WHY（原因）：为什么在这个场合？事件的背景/目的 6. HOW（方式）：通过什么方式记录？照片呈现了什么氛围/状态 返回JSON格式： {{ "who": ["人物1", "人物2"], "what": "事件描述", "when": "时间（保持原始精度）", "where": "地点描述", "why": "事件原因/背景", "how": "记录方式/氛围", "topic": "主题分类（event/travel/work/social等）", "confidence": 0.8 }} 示例： 如果照片显示在公司开会，可能是： {{"who": ["张三", "李四"], "what": "产品评审会议", "when": "2026-09-20", "where": "北京办公室会议室", "why": "讨论Q4产品计划", "how": "会议场景，多人参与", "topic": "work", "confidence": 0.85}} 只输出JSON，不要解释。如无明确事件信息返回空对象{{}}。""" # 调用LLM API async with aiohttp.ClientSession() as session: payload = { "model": config.llm.get("model", "agnes-2.5-flash"), "messages": [{"role": "user", "content": prompt}], "max_tokens": 300 } async with session.post( f"{config.get('openai.base_url', 'https://api.agnes-ai.cn/v1')}/chat/completions", json=payload, headers={"Authorization": f"Bearer {config.get('openai.api_key', '')}"} ) as resp: if resp.status == 200: result = await resp.json() text = result.get("choices", [{}])[0].get("message", {}).get("content", "") import re json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: try: event_data = json.loads(json_match.group()) if event_data: # 非空对象才记录 topic = event_data.get("topic", "event") who = event_data.get("who", []) what = event_data.get("what", "") when = event_data.get("when", "") where = event_data.get("where", "") why = event_data.get("why", "") how = event_data.get("how", "") confidence = event_data.get("confidence", 0.7) # 提取关键词作为事实 if what: self.agent.profile_manager.add_fact( subject_id="visual_memory", predicate="attended_event", object=what, topic=topic, confidence=confidence ) # 如果有时空信息，记录为事件 if what and (when or where): # 构建完整事件描述 description = f"{what}" if where: description += f"于{where}" if when: description += f"（{when}）" if why: description += f"，原因是{why}" # 添加事件 self.agent.profile_manager.add_event( timestamp=time.time(), participant_ids=who if who else ["visual_memory"], event_type=topic, description=description, importance=min(confidence + 0.2, 1.0), # 事件重要性略高于事实 metadata={ "scene_description": scene_description, "exif": exif, "face_count": len(face_snapshots), "source_url": source_url, "raw_who": who, "raw_where": where, "raw_when": when, "raw_why": why, "raw_how": how } ) logger.info(f"📸 记录事件: topic={topic}, what={what[:30]}..., where={where[:20] if where else 'unknown'}") except Exception as e: logger.debug(f"事件解析失败: {e}") except Exception as e: logger.debug(f"照片事件提取失败: {e}") async def _encode_image(self, img_bytes: bytes) -> Optional[List[float]]: """调用WeMM API编码图像""" import aiohttp async with aiohttp.ClientSession() as session: data = aiohttp.FormData() data.add_field('file', img_bytes, filename='frame.jpg', content_type='image/jpeg') data.add_field('dimension', str(self.wemm_dimension)) try: async with session.post(f"{self.wemm_url}/embed", data=data) as resp: if resp.status == 200: result = await resp.json() embeddings = result.get("embeddings", []) if embeddings: return embeddings[0] except Exception as e: logger.error(f"WeMM编码失败: {e}") return None async def search_memories(self, query_text: str, top_k: int = 3) -> List[Dict]: """搜索视觉记忆""" import numpy as np import aiohttp # 编码查询文本 query_emb = await self._encode_text(query_text) if not query_emb: return [] query_vec = np.array(query_emb) # 计算相似度 results = [] for mem in self._visual_memories: mem_vec = np.array(mem["embedding"]) similarity = float(np.dot(query_vec, mem_vec)) # L2归一化，点积即余弦相似度 results.append({ "similarity": similarity, "timestamp": mem["timestamp"], "frame_hash": mem["frame_hash"] }) # 按相似度排序 results.sort(key=lambda x: x["similarity"], reverse=True) return results[:top_k] async def _encode_text(self, text: str) -> Optional[List[float]]: """调用WeMM API编码文本""" import aiohttp async with aiohttp.ClientSession() as session: try: async with session.post( f"{self.wemm_url}/embed", json={"inputs": text, "dimension": self.wemm_dimension} ) as resp: if resp.status == 200: result = await resp.json() embeddings = result.get("embeddings", []) if embeddings: return embeddings[0] except Exception as e: logger.error(f"WeMM文本编码失败: {e}") return None async def get_recent_summary(self, last_minutes: int = 5) -> str: """获取最近N分钟的视觉摘要""" import numpy as np cutoff_time = time.time() - (last_minutes * 60) recent = [m for m in self._visual_memories if m["timestamp"] > cutoff_time] if not recent: return "过去{}分钟没有新的视觉记录".format(last_minutes) # 简单摘要：返回帧哈希和时间 summaries = [] for mem in recent[-5:]: # 最近5条 from datetime import datetime ts = datetime.fromtimestamp(mem["timestamp"]).strftime("%H:%M:%S") summaries.append(f"- {ts} (帧#{mem['frame_hash']})") return "最近视觉记录:\n" + "\n".join(summaries) def get_memory_stats(self) -> Dict: """获取记忆统计""" return { "total_memories": len(self._visual_memories), "max_entries": self.max_entries, "memory_enabled": self.memory_enabled, "last_learn": self._last_learn_time } # ========== LiveKit 集成 ========== async def entrypoint(ctx: JobContext): logger.info("=" * 60) logger.info("🚀 LiveKit Voice AI Agent v5 (配置驱动版)") logger.info(f" 房间: {ctx.room.name}") logger.info(f" LLM: {config.llm.get('backend')} / {config.llm.get('model')}") logger.info(f" STT: {config.stt.get('backend')}") logger.info(f" TTS: {config.tts.get('backend')}") logger.info("=" * 60) agent = VoiceAgent() await agent.initialize() await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_AND_VIDEO) sample_rate = int(config.pipeline.get("sample_rate", 24000)) source = rtc.AudioSource(sample_rate, 1) audio_track = rtc.LocalAudioTrack.create_audio_track("agent_voice", source) options = rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE) await ctx.room.local_participant.publish_track(audio_track, options) agent._audio_source = source # 视频分析器初始化 if config.get("vision.enabled", False): agent.visual_monitor = VisualMonitor(agent) await agent.visual_monitor.initialize() @ctx.room.on("participant_connected") async def on_participant_connected(participant: rtc.RemoteParticipant): logger.info(f"👤 新参与者: {participant.identity}") # 注册到对话记录器 if hasattr(agent, 'dialogue_logger') and agent.dialogue_logger._conn: agent.dialogue_logger._conn.execute(''' INSERT OR IGNORE INTO speakers (identity, first_seen, last_seen, total_turns) VALUES (?, ?, ?, 0) ''', (participant.identity, time.time(), time.time())) agent.dialogue_logger._conn.commit() # 注册到人物档案管理器 if hasattr(agent, 'profile_manager'): agent.profile_manager.register_profile( identity=participant.identity, name=participant.identity, # 初始用 identity 作为名字 preferences={"joined_at": time.time()} ) @ctx.room.on("track_subscribed") async def on_track_subscribed(track: rtc.Track, publication: rtc.TrackPublication, participant: rtc.RemoteParticipant): if track.kind == rtc.TrackKind.KIND_AUDIO: logger.info(f"🎵 订阅音频: {participant.identity}") asyncio.create_task(_process_audio_stream(agent, track)) elif track.kind == rtc.TrackKind.KIND_VIDEO: logger.info(f"📹 订阅视频: {participant.identity}") asyncio.create_task(_process_video_stream(agent, track)) logger.info("✅ Agent v5 已就绪") while ctx.room.isconnected: await asyncio.sleep(1) await agent.close() async def _process_audio_stream(agent: VoiceAgent, track: rtc.Track): stream = rtc.AudioStream(track) async for event in stream: if hasattr(event, 'frame'): frame = event.frame audio_bytes = bytes(frame.buffer) await agent.process_audio_frame(audio_bytes) async def _process_video_stream(agent: VoiceAgent, track: rtc.Track): """处理视频轨道，用于主动监控""" if not hasattr(agent, 'visual_monitor'): return stream = rtc.VideoStream(track) async for frame in stream: await agent.visual_monitor.analyze_frame(frame) # ========== 命令行入口 ========== if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="LiveKit Voice AI Agent v5") parser.add_argument("action", choices=["start", "test", "config"]) args = parser.parse_args() if args.action == "start": from livekit.agents import WorkerOptions from livekit.agents.worker import AgentServer options = WorkerOptions( entrypoint_fnc=entrypoint, agent_name="voice-ai-agent-v5", ws_url=config.livekit.get("url", "http://localhost:7880"), api_key=config.livekit.get("api_key", "devkey"), api_secret=config.livekit.get("api_secret", "secret"), ) server = AgentServer.from_server_options(options) asyncio.run(server.run(devmode=True)) elif args.action == "test": print("🧪 Agent v5 测试") print(f"STT: {config.stt.get('endpoint')}") print(f"LLM: {config.llm.get('backend')} / {config.llm.get('model')}") print(f"TTS: {config.tts.get('backend')} / {config.tts.get('voice')}") print(f"VAD: {config.vad.get('backend')} (threshold={config.vad.get('threshold')})") print() # 测试 STT try: import httpx, io, wave async def test_stt(): async with httpx.AsyncClient(base_url=config.stt.get("endpoint", "http://localhost:8080"), timeout=10.0) as client: buf = io.BytesIO() with wave.open(buf, "w") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(16000) wf.writeframes(b"\x00\x00" * 16000) resp = await client.post("/stream", files={"file": ("test.wav", buf.getvalue(), "audio/wav")}) print(f"✅ STT: {resp.json().get('text', 'N/A')}") asyncio.run(test_stt()) except Exception as e: print(f"❌ STT 失败: {e}") # 测试 LLM try: from openai import AsyncOpenAI async def test_llm(): client = AsyncOpenAI( api_key=config.get("openai.api_key", ""), base_url=config.get("openai.base_url", "https://api.agnes-ai.cn/v1") ) resp = await client.chat.completions.create( model=config.llm.get("model", "agnes-2.5-flash"), messages=[{"role": "user", "content": "你好"}], max_tokens=50, ) print(f"✅ LLM: {resp.choices[0].message.content}") asyncio.run(test_llm()) except Exception as e: print(f"❌ LLM 失败: {e}") # 显示配置 print() print("📋 当前配置:") print(f" Sample Rate: {config.pipeline.get('sample_rate')}Hz") print(f" Chunk Size: {config.pipeline.get('chunk_size_ms')}ms") print(f" Memory Limit: {config.memory.get('token_limit')} tokens") elif args.action == "config": print("📋 完整配置:") # 脱敏打印，隐藏 API Key safe_data = config.data.copy() for section in ["livekit", "openai", "funasr"]: if section in safe_data and isinstance(safe_data[section], dict): for key in list(safe_data[section].keys()): if "secret" in key.lower() or "key" in key.lower(): safe_data[section][key] = "***REDACTED***" print(yaml.dump(safe_data, default_flow_style=False))