feat: add emotion prompt

.env.example

@@ -23,6 +23,10 @@ CUSTOM_LLM_BASE_URL=http://localhost/v1
 CUSTOM_LLM_MODEL=Qwen-VL
 CUSTOM_LLM_API_KEY=
 CUSTOM_LLM_VERIFY_SSL=false
+CUSTOM_SAVE_MODEL_IMAGES=false
+
+# CUSTOM_TEXT_LLM_MODEL=
+# CUSTOM_VISION_LLM_MODEL=
 
 # CUSTOM_LLM_BASE_URL=https://api.deepseek.com
 # CUSTOM_LLM_MODEL=deepseek-v4-flash
@@ -31,7 +35,7 @@ CUSTOM_LLM_VERIFY_SSL=false
 
 
 # TTS blackbox
-CUSTOM_TTS_URL=http://localhost:5000/tts-blackbox
+CUSTOM_TTS_URL=http://localhost:5050/tts-blackbox
 CUSTOM_TTS_MODEL=voxcpmtts
 # CUSTOM_TTS_PROMPT_WAV=/home/verachen/Workspace/livekit/agents/2food.wav
 CUSTOM_TTS_STREAMING=true

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+.env
+model_images/

custom_agent.py

@@ -1,7 +1,11 @@
+import base64
+import json
 import logging
 import os
+import re
 import time
 from collections.abc import AsyncIterable
+from dataclasses import dataclass
 from pathlib import Path
 
 from dotenv import load_dotenv
@@ -54,14 +58,95 @@ GENERAL_INSTRUCTIONS = """
 回答自然、简洁、准确。
 """.strip()
 
+EMOTION_INSTRUCTIONS = """
+每次回复必须先输出一个情绪标签，格式严格为：<emotion=neutral>
+emotion 只能从 neutral、happy、sad、angry、surprised、fearful、calm、concerned 中选择。
+情绪标签之后直接输出给用户的正常回复，不要解释标签。
+""".strip()
+
 ROOM_LOCATOR_MODE = "room_locator"
 GENERAL_MODE = "general"
+VOICE_INPUT_MODE = "voice"
+VISION_VOICE_INPUT_MODE = "vision_voice"
+AUTO_INPUT_MODE = "auto"
+VISION_FRAME_TOPIC = "vision.frame"
+
+DEFAULT_EMOTION = "neutral"
+EMOTION_LABELS = {
+    "neutral",
+    "happy",
+    "sad",
+    "angry",
+    "surprised",
+    "fearful",
+    "calm",
+    "concerned",
+}
+EMOTION_PREFIX_RE = re.compile(r"^\s*<emotion=([a-z_]+)>\s*", re.IGNORECASE)
+TTS_EMOTION_MARKUP_RE = re.compile(r"<\s*emotion\s*=\s*[^>]{1,80}>\s*", re.IGNORECASE)
+TTS_EMOTION_LINE_RE = re.compile(
+    r"^\s*(?:emotion|情绪)\s*[:：=]\s*[\w\u4e00-\u9fff-]{1,40}\s*[,，。.!！\s-]*",
+    re.IGNORECASE,
+)
+MAX_EMOTION_PREFIX_CHARS = 80
+
+
+@dataclass
+class VisionFrame:
+    image_data_url: str
+    received_at: float
+    mime_type: str
+    saved_path: str | None = None
+
+
+class VisionFrameStore:
+    def __init__(self, *, max_age_seconds: float) -> None:
+        self._max_age_seconds = max_age_seconds
+        self._latest_frame: VisionFrame | None = None
+
+    def update(self, *, image: str, mime_type: str, saved_path: str | None = None) -> None:
+        self._latest_frame = VisionFrame(
+            image_data_url=f"data:{mime_type};base64,{image}",
+            received_at=time.monotonic(),
+            mime_type=mime_type,
+            saved_path=saved_path,
+        )
+
+    def consume_fresh(self) -> VisionFrame | None:
+        frame = self._latest_frame
+        if frame is None:
+            return None
+
+        age = time.monotonic() - frame.received_at
+        self._latest_frame = None
+        if age > self._max_age_seconds:
+            logger.info("Dropping stale vision frame: age=%.3fs", age)
+            return None
+
+        return frame
 
 
 class CustomAgent(Agent):
-    def __init__(self, *, memory_client: MemoryRecallClient | None = None) -> None:
-        super().__init__(instructions=GENERAL_INSTRUCTIONS)
+    def __init__(
+        self,
+        *,
+        memory_client: MemoryRecallClient | None = None,
+        vision_store: VisionFrameStore | None = None,
+        input_mode: str = AUTO_INPUT_MODE,
+        text_llm: llm.LLM | None = None,
+        vision_llm: llm.LLM | None = None,
+        model_image_save_dir: Path | None = None,
+    ) -> None:
+        super().__init__(instructions=_with_emotion_instructions(GENERAL_INSTRUCTIONS))
         self._memory_client = memory_client
+        self._vision_store = vision_store
+        self._input_mode = input_mode
+        self._text_llm = text_llm
+        self._vision_llm = vision_llm
+        self._model_image_save_dir = model_image_save_dir
+        self.current_emotion = DEFAULT_EMOTION
+        self._emotion_prefix_buffer = ""
+        self._emotion_prefix_done = True
 
     async def on_enter(self) -> None:
         # self.session.generate_reply(instructions="greet the user and introduce yourself")
@@ -77,14 +162,20 @@ class CustomAgent(Agent):
 
         user_query = _latest_user_text(chat_ctx)
         mode = _select_mode(user_query)
-        logger.info("Selected agent mode: %s", mode)
+        vision_frame = self._consume_vision_frame()
+        logger.info(
+            "Selected agent mode: %s input_mode=%s has_image=%s",
+            mode,
+            self._input_mode,
+            vision_frame is not None,
+        )
 
         chat_ctx = chat_ctx.copy()
         update_chat_instructions(
             chat_ctx,
-            instructions=ROOM_LOCATOR_INSTRUCTIONS
-            if mode == ROOM_LOCATOR_MODE
-            else GENERAL_INSTRUCTIONS,
+            instructions=_with_emotion_instructions(
+                ROOM_LOCATOR_INSTRUCTIONS if mode == ROOM_LOCATOR_MODE else GENERAL_INSTRUCTIONS
+            ),
             add_if_missing=True,
         )
 
@@ -93,7 +184,16 @@ class CustomAgent(Agent):
             if memory_context:
                 chat_ctx = _with_memory_as_latest_user_message(chat_ctx, memory_context)
 
-        llm_result = Agent.default.llm_node(self, chat_ctx, tools, model_settings)
+        if vision_frame is not None:
+            self._save_model_vision_frame(vision_frame)
+            chat_ctx = _with_vision_as_latest_user_message(chat_ctx, vision_frame)
+
+        llm_result = self._run_selected_llm(
+            chat_ctx,
+            tools,
+            model_settings,
+            has_image=vision_frame is not None,
+        )
         if not hasattr(llm_result, "__aiter__"):
             elapsed = time.perf_counter() - llm_node_started_at
             logger.info("LLM node completed without streaming in %.3fs", elapsed)
@@ -102,6 +202,8 @@ class CustomAgent(Agent):
         async def _instrumented_stream() -> AsyncIterable[llm.ChatChunk | str | FlushSentinel]:
             first_chunk_at: float | None = None
             chunk_count = 0
+            self._emotion_prefix_buffer = ""
+            self._emotion_prefix_done = False
             try:
                 async for chunk in llm_result:
                     chunk_count += 1
@@ -111,7 +213,8 @@ class CustomAgent(Agent):
                             "LLM first chunk after %.3fs",
                             first_chunk_at - llm_node_started_at,
                         )
-                    yield chunk
+                    async for output_chunk in self._observe_emotion_prefix(chunk):
+                        yield output_chunk
             finally:
                 finished_at = time.perf_counter()
                 logger.info(
@@ -123,6 +226,116 @@ class CustomAgent(Agent):
 
         return _instrumented_stream()
 
+    def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
+        return Agent.default.tts_node(self, _strip_emotion_for_tts(text), model_settings)
+
+    def _consume_vision_frame(self) -> VisionFrame | None:
+        if self._input_mode == VOICE_INPUT_MODE or self._vision_store is None:
+            return None
+        return self._vision_store.consume_fresh()
+
+    def _save_model_vision_frame(self, vision_frame: VisionFrame) -> None:
+        if self._model_image_save_dir is None:
+            return
+
+        try:
+            _, b64_data = vision_frame.image_data_url.split(",", 1)
+            image_bytes = base64.b64decode(b64_data, validate=True)
+        except Exception:
+            logger.exception("Failed to decode model vision frame for debug save")
+            return
+
+        extension = _image_extension_from_mime_type(vision_frame.mime_type)
+        timestamp_ms = int(time.time() * 1000)
+        path = self._model_image_save_dir / f"{timestamp_ms}_model_input{extension}"
+
+        try:
+            self._model_image_save_dir.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(image_bytes)
+        except Exception:
+            logger.exception("Failed to save model vision frame: path=%s", path)
+            return
+
+        logger.info(
+            "Saved model vision frame: path=%s bytes=%s source_path=%s",
+            path,
+            len(image_bytes),
+            vision_frame.saved_path,
+        )
+
+    def _run_selected_llm(
+        self,
+        chat_ctx: ChatContext,
+        tools: list[llm.Tool],
+        model_settings: ModelSettings,
+        *,
+        has_image: bool,
+    ) -> AsyncIterable[llm.ChatChunk | str | FlushSentinel]:
+        selected_llm = self._vision_llm if has_image else self._text_llm
+        if selected_llm is None:
+            return Agent.default.llm_node(self, chat_ctx, tools, model_settings)
+
+        activity = self._get_activity_or_raise()
+        tool_choice = model_settings.tool_choice
+        conn_options = activity.session.conn_options.llm_conn_options
+
+        async def _stream() -> AsyncIterable[llm.ChatChunk | str | FlushSentinel]:
+            async with selected_llm.chat(
+                chat_ctx=chat_ctx,
+                tools=tools,
+                tool_choice=tool_choice,
+                conn_options=conn_options,
+            ) as stream:
+                async for chunk in stream:
+                    yield chunk
+
+        return _stream()
+    async def _observe_emotion_prefix(
+        self, chunk: llm.ChatChunk | str | FlushSentinel
+    ) -> AsyncIterable[llm.ChatChunk | str | FlushSentinel]:
+        if isinstance(chunk, str):
+            self._consume_emotion_prefix(chunk)
+            yield chunk
+            return
+
+        if isinstance(chunk, llm.ChatChunk) and chunk.delta and chunk.delta.content:
+            self._consume_emotion_prefix(chunk.delta.content)
+            yield chunk
+            return
+
+        yield chunk
+
+    def _consume_emotion_prefix(self, content: str) -> None:
+        if self._emotion_prefix_done:
+            return
+
+        self._emotion_prefix_buffer += content
+        match = EMOTION_PREFIX_RE.match(self._emotion_prefix_buffer)
+        if match:
+            emotion = match.group(1).lower()
+            if emotion not in EMOTION_LABELS:
+                logger.warning("LLM returned unsupported emotion=%s, using neutral", emotion)
+                emotion = DEFAULT_EMOTION
+
+            self.current_emotion = emotion
+            self._emotion_prefix_done = True
+            self._emotion_prefix_buffer = ""
+            logger.info("LLM emotion selected: %s", emotion)
+            return
+
+        candidate = self._emotion_prefix_buffer.lstrip().lower()
+        might_still_be_prefix = (
+            not candidate
+            or "<emotion=".startswith(candidate)
+            or (candidate.startswith("<emotion=") and ">" not in candidate)
+        )
+        if might_still_be_prefix and len(candidate) <= MAX_EMOTION_PREFIX_CHARS:
+            return
+
+        self._emotion_prefix_done = True
+        self._emotion_prefix_buffer = ""
+        logger.warning("LLM response did not start with an emotion prefix")
+
     async def _recall_room_memory(self, chat_ctx: ChatContext) -> str:
         if self._memory_client is None:
             return ""
@@ -161,6 +374,69 @@ def _select_mode(user_query: str) -> str:
     return GENERAL_MODE
 
 
+def _with_emotion_instructions(instructions: str) -> str:
+    return f"{instructions}\n\n{EMOTION_INSTRUCTIONS}"
+
+
+async def _strip_emotion_for_tts(text: AsyncIterable[str]) -> AsyncIterable[str]:
+    prefix_buffer = ""
+    scanning_prefix = True
+
+    async for chunk in text:
+        if not chunk:
+            continue
+
+        if scanning_prefix:
+            prefix_buffer += chunk
+            cleaned, done = _strip_leading_tts_emotion(prefix_buffer)
+            if not done:
+                continue
+
+            scanning_prefix = False
+            prefix_buffer = ""
+            if cleaned:
+                yield _strip_inline_tts_emotion(cleaned)
+            continue
+
+        cleaned = _strip_inline_tts_emotion(chunk)
+        if cleaned:
+            yield cleaned
+
+    if scanning_prefix and prefix_buffer:
+        cleaned, _ = _strip_leading_tts_emotion(prefix_buffer, force=True)
+        cleaned = _strip_inline_tts_emotion(cleaned)
+        if cleaned:
+            yield cleaned
+
+
+def _strip_leading_tts_emotion(text: str, *, force: bool = False) -> tuple[str, bool]:
+    match = TTS_EMOTION_MARKUP_RE.match(text)
+    if match:
+        return text[match.end() :], True
+
+    match = TTS_EMOTION_LINE_RE.match(text)
+    if match:
+        return text[match.end() :], True
+
+    candidate = text.lstrip().lower()
+    might_still_be_emotion = (
+        not candidate
+        or "<emotion=".startswith(candidate)
+        or (candidate.startswith("<emotion") and ">" not in candidate)
+        or "emotion".startswith(candidate)
+        or (candidate.startswith("emotion") and len(candidate) <= MAX_EMOTION_PREFIX_CHARS)
+        or "情绪".startswith(candidate)
+    )
+    if not force and might_still_be_emotion and len(candidate) <= MAX_EMOTION_PREFIX_CHARS:
+        return "", False
+
+    return text, True
+
+
+def _strip_inline_tts_emotion(text: str) -> str:
+    return TTS_EMOTION_MARKUP_RE.sub("", text)
+
+
 def _is_room_locator_query(normalized_text: str) -> bool:
     room_context_hints = (
         "房间",
@@ -269,6 +545,73 @@ def _with_memory_as_latest_user_message(chat_ctx: ChatContext, memory_context: s
     return chat_ctx
 
 
+def _with_vision_as_latest_user_message(chat_ctx: ChatContext, vision_frame: VisionFrame) -> ChatContext:
+    chat_ctx = chat_ctx.copy()
+    image_content = llm.ImageContent(
+        image=vision_frame.image_data_url,
+        mime_type=vision_frame.mime_type,
+        inference_detail="auto",
+    )
+
+    for index in range(len(chat_ctx.items) - 1, -1, -1):
+        item = chat_ctx.items[index]
+        if isinstance(item, ChatMessage) and item.role == "user":
+            user_msg = item.model_copy(deep=True)
+            content = list(user_msg.content)
+            content.append(image_content)
+            user_msg.content = content
+            chat_ctx.items[index] = user_msg
+            return chat_ctx
+
+    chat_ctx.items.append(ChatMessage(role="user", content=[image_content]))
+    return chat_ctx
+
+
+def _normalize_input_mode(value: str | None) -> str:
+    if not value:
+        return AUTO_INPUT_MODE
+
+    normalized = value.strip().lower().replace("-", "_")
+    aliases = {
+        "image_voice": VISION_VOICE_INPUT_MODE,
+        "image": VISION_VOICE_INPUT_MODE,
+        "vision": VISION_VOICE_INPUT_MODE,
+        "vision_voice": VISION_VOICE_INPUT_MODE,
+        "voice_image": VISION_VOICE_INPUT_MODE,
+        "audio": VOICE_INPUT_MODE,
+        "voice": VOICE_INPUT_MODE,
+        "auto": AUTO_INPUT_MODE,
+    }
+    mode = aliases.get(normalized)
+    if mode is not None:
+        return mode
+
+    logger.warning("Invalid CUSTOM_AGENT_INPUT_MODE=%r, using %s", value, AUTO_INPUT_MODE)
+    return AUTO_INPUT_MODE
+
+
+def _image_extension_from_mime_type(mime_type: str) -> str:
+    normalized = mime_type.strip().lower()
+    if normalized == "image/png":
+        return ".png"
+    if normalized == "image/webp":
+        return ".webp"
+    if normalized == "image/gif":
+        return ".gif"
+    return ".jpg"
+
+
+def _model_image_save_dir_from_env() -> Path | None:
+    if not _env_bool("CUSTOM_SAVE_MODEL_IMAGES", True):
+        return None
+
+    configured = os.getenv("CUSTOM_MODEL_IMAGE_SAVE_DIR")
+    if configured:
+        return Path(configured).expanduser()
+
+    return Path(__file__).with_name("model_images")
+
+
 server = AgentServer()
 
 
@@ -295,8 +638,12 @@ async def entrypoint(ctx: JobContext) -> None:
     LLM_BASE_URL = os.getenv("CUSTOM_LLM_BASE_URL")
     LLM_MODEL = os.getenv("CUSTOM_LLM_MODEL", "qwen-max")
     LLM_API_KEY = os.getenv("CUSTOM_LLM_API_KEY")
+    TEXT_LLM_MODEL = os.getenv("CUSTOM_TEXT_LLM_MODEL", LLM_MODEL)
+    VISION_LLM_MODEL = os.getenv("CUSTOM_VISION_LLM_MODEL", LLM_MODEL)
+    INPUT_MODE = _normalize_input_mode(os.getenv("CUSTOM_AGENT_INPUT_MODE"))
     if not LLM_API_KEY:
         raise RuntimeError(f"CUSTOM_LLM_API_KEY is not set in {CUSTOM_ENV_PATH}")
+    logger.info("Using LLM model=%s base_url=%s", LLM_MODEL, LLM_BASE_URL or "OpenAI default")
 
     TTS_URL = os.getenv("CUSTOM_TTS_URL") or os.getenv(
         "VOXCPM_TTS_URL", "http://localhost:5000/tts-blackbox"
@@ -339,14 +686,29 @@ async def entrypoint(ctx: JobContext) -> None:
             http_client=http_client,
         )
 
+    base_llm = openai.LLM(
+        model=LLM_MODEL,
+        client=openai_client,
+    )
+    text_llm = (
+        openai.LLM(model=TEXT_LLM_MODEL, client=openai_client)
+        if TEXT_LLM_MODEL != LLM_MODEL
+        else base_llm
+    )
+    vision_llm = (
+        openai.LLM(model=VISION_LLM_MODEL, client=openai_client)
+        if VISION_LLM_MODEL != LLM_MODEL
+        else base_llm
+    )
+    vision_store = VisionFrameStore(
+        max_age_seconds=_env_float("CUSTOM_VISION_FRAME_MAX_AGE_SECONDS", 8.0)
+    )
+
     session: AgentSession = AgentSession(
         # 1. Custom ASR blackbox with StreamAdapter
         stt=stt_stream,
         # 2. OpenAI-compatible LLM, e.g. MiniMax, Qwen, or OpenAI.
-        llm=openai.LLM(
-            model=LLM_MODEL,
-            client=openai_client,
-        ),
+        llm=base_llm,
         # 3. TTS blackbox
         tts=BlackboxTTS(
             url=TTS_URL,
@@ -388,6 +750,47 @@ async def entrypoint(ctx: JobContext) -> None:
         elif item.role == "assistant" and item.metrics:
             logger.info("Assistant turn metrics: %s", item.metrics)
 
+    @ctx.room.on("data_received")
+    def _on_data_received(data_packet) -> None:
+        packet_topic = getattr(data_packet, "topic", None)
+        if packet_topic not in {None, "", VISION_FRAME_TOPIC}:
+            return
+
+        if INPUT_MODE == VOICE_INPUT_MODE:
+            logger.info("Ignoring vision frame because CUSTOM_AGENT_INPUT_MODE=%s", INPUT_MODE)
+            return
+
+        try:
+            payload = json.loads(data_packet.data.decode("utf-8"))
+        except Exception:
+            logger.exception("Failed to decode vision frame payload")
+            return
+
+        if payload.get("type") != "vision_frame" and payload.get("topic") != VISION_FRAME_TOPIC:
+            return
+
+        image = payload.get("image")
+        if not isinstance(image, str) or not image:
+            logger.warning("Received vision frame without image data")
+            return
+
+        mime_type = payload.get("mime_type")
+        if not isinstance(mime_type, str) or not mime_type:
+            mime_type = "image/jpeg"
+
+        saved_path = payload.get("saved_path")
+        vision_store.update(
+            image=image,
+            mime_type=mime_type,
+            saved_path=saved_path if isinstance(saved_path, str) else None,
+        )
+        logger.info(
+            "Cached vision frame: mime_type=%s image_chars=%s saved_path=%s",
+            mime_type,
+            len(image),
+            saved_path,
+        )
+
     memory_client = (
         MemoryRecallClient(
             url=MEMORY_URL,
@@ -400,7 +803,14 @@ async def entrypoint(ctx: JobContext) -> None:
     )
 
     await session.start(
-        agent=CustomAgent(memory_client=memory_client),
+        agent=CustomAgent(
+            memory_client=memory_client,
+            vision_store=vision_store,
+            input_mode=INPUT_MODE,
+            text_llm=text_llm,
+            vision_llm=vision_llm,
+            model_image_save_dir=_model_image_save_dir_from_env(),
+        ),
         room=ctx.room,
         room_options=room_io.RoomOptions(
             audio_output=room_io.AudioOutputOptions(