feat: update stream_agent_raw function

hyacinthus · web-flow · commit a1072c8ff612 · 2025-10-19T19:56:01.000+08:00
Merged PR #863 with updates to stream_agent_raw function
diff --git a/app/services/tg/bot/kind/ai_relayer/router.py b/app/services/tg/bot/kind/ai_relayer/router.py
@@ -1,5 +1,8 @@
 import inspect
+import json
 import logging
+from datetime import datetime
+from typing import List
 
 import telegramify_markdown
 from aiogram import Router
@@ -9,6 +12,8 @@
 
 from intentkit.core.client import execute_agent
 from intentkit.models.chat import AuthorType, ChatMessageCreate
+from intentkit.models.redis import get_redis
+from intentkit.models.user import User
 from intentkit.utils.slack_alert import send_slack_message
 
 from app.services.tg.bot import pool
@@ -20,6 +25,133 @@
 
 logger = logging.getLogger(__name__)
 
+# Cache configuration
+CACHE_TTL = 24 * 60 * 60  # 1 day in seconds
+MAX_CACHED_MESSAGES = 10
+MAX_CACHE_SIZE_BYTES = 2048  # 2KB
+
+
+async def cache_message(chat_id: int, agent_id: str, message: Message) -> None:
+    """Cache a message in Redis for context retrieval.
+
+    Args:
+        chat_id: The chat ID where the message was sent
+        agent_id: The agent ID to identify the agent
+        message: The message to cache
+    """
+    try:
+        redis = get_redis()
+        cache_key = f"intentkit:tg_context:{agent_id}:{chat_id}"
+
+        # Prepare message data for caching
+        username = (
+            message.from_user.username
+            if message.from_user and message.from_user.username
+            else "Unknown"
+        )
+
+        message_data = {
+            "text": message.text,
+            "username": username,
+            "timestamp": datetime.now().isoformat(),
+            "message_id": message.message_id,
+        }
+
+        # Add to Redis list (right push to maintain chronological order - oldest first)
+        await redis.rpush(cache_key, json.dumps(message_data))
+
+        # Trim list to keep only the latest MAX_CACHED_MESSAGES (remove from left)
+        await redis.ltrim(cache_key, -MAX_CACHED_MESSAGES, -1)
+
+        # Set expiration
+        await redis.expire(cache_key, CACHE_TTL)
+
+    except Exception as e:
+        logger.warning(f"Failed to cache message: {e}")
+
+
+async def get_cached_context(chat_id: int, agent_id: str) -> List[dict]:
+    """Retrieve cached messages for context.
+
+    Args:
+        chat_id: The chat ID to get context for
+        agent_id: The agent ID to identify the agent
+
+    Returns:
+        List of cached message data, limited by size and count
+    """
+    try:
+        redis = get_redis()
+        cache_key = f"intentkit:tg_context:{agent_id}:{chat_id}"
+
+        # Get all cached messages (oldest first due to rpush)
+        cached_messages = await redis.lrange(cache_key, 0, -1)
+
+        if not cached_messages:
+            return []
+
+        # Parse messages and apply size limit
+        context_messages = []
+        total_size = 0
+
+        for msg_json in cached_messages:
+            try:
+                msg_data = json.loads(msg_json)
+                msg_size = len(json.dumps(msg_data))
+
+                if total_size + msg_size > MAX_CACHE_SIZE_BYTES:
+                    break
+
+                context_messages.append(msg_data)
+                total_size += msg_size
+
+            except json.JSONDecodeError:
+                continue
+
+        return context_messages
+
+    except Exception as e:
+        logger.warning(f"Failed to get cached context: {e}")
+        return []
+
+
+async def clear_cached_context(chat_id: int, agent_id: str) -> None:
+    """Clear cached messages after agent processes them.
+
+    Args:
+        chat_id: The chat ID to clear context for
+        agent_id: The agent ID to identify the agent
+    """
+    try:
+        redis = get_redis()
+        cache_key = f"intentkit:tg_context:{agent_id}:{chat_id}"
+        await redis.delete(cache_key)
+
+    except Exception as e:
+        logger.warning(f"Failed to clear cached context: {e}")
+
+
+async def get_user_id(from_user) -> str:
+    """
+    Extract user_id from telegram message from_user.
+
+    Args:
+        from_user: Telegram user object from message.from_user
+
+    Returns:
+        str: User ID, either from database lookup or fallback to username/user_id
+    """
+    if from_user and from_user.username:
+        user = await User.get_by_tg(from_user.username)
+        if user:
+            return user.id
+        else:
+            return from_user.username
+    elif from_user:
+        return str(from_user.id)
+    else:
+        raise ValueError("No valid user information available")
+
 
 def cur_func_name():
     return inspect.stack()[1][3]
@@ -67,25 +199,43 @@ async def gp_command_start(message: Message):
 )
 async def gp_process_message(message: Message) -> None:
     bot = await message.bot.get_me()
-    if (
+
+    # Check if this message should trigger a bot reply
+    should_reply = (
         message.reply_to_message
         and message.reply_to_message.from_user.id == message.bot.id
-    ) or bot.username in message.text:
+    ) or bot.username in message.text
+
+    if should_reply:
         cached_bot_item = pool.bot_by_token(message.bot.token)
         if cached_bot_item is None:
             logger.warning(f"bot with token {message.bot.token} not found in cache.")
             return
 
-        if message.from_user:
+        try:
+            user_id = await get_user_id(message.from_user)
+            is_owner = (
+                cached_bot_item._owner == message.from_user.username
+                if message.from_user
+                else False
+            )
             logger.info(f"message from: {message.from_user}")
-            is_owner = cached_bot_item._owner == message.from_user.username
-        else:
+        except ValueError:
             is_owner = False
+            logger.error(
+                f"telegram message from user without username: {message.from_user}"
+            )
+            return
 
         # Add processing reaction
         await message.react([ReactionTypeEmoji(emoji="🤔")])
 
         try:
+            # Get cached context messages
+            context_messages = await get_cached_context(
+                message.chat.id, cached_bot_item.agent_id
+            )
+
             # remove bot name tag from text
             message_text = remove_bot_name(bot.username, message.text)
             if len(message_text) > 65535:
@@ -95,24 +245,40 @@ async def gp_process_message(message: Message) -> None:
                         f"length: {len(message_text)}\n"
                         f"chat_id:{message.chat.id}\n"
                         f"agent:{cached_bot_item.agent_id}\n"
-                        f"user:{message.from_user.id}\n"
+                        f"user:{user_id}\n"
                         f"content:{message_text[:100]}..."
                     )
                 )
 
+            # Wrap message with group context and username
+            username = (
+                message.from_user.username
+                if message.from_user and message.from_user.username
+                else "Unknown"
+            )
+
+            # Build message with context
+            if context_messages:
+                context_text = "Recent conversation context:\n"
+                # Messages are already in chronological order (oldest first)
+                for ctx_msg in context_messages:
+                    context_text += f"@{ctx_msg['username']}: {ctx_msg['text']}\n"
+                context_text += f"\nCurrent message from @{username}: {message_text}"
+                wrapped_message = f"[Group Message with Context]: {context_text}"
+            else:
+                wrapped_message = f"[Group Message from @{username}]: {message_text}"
+
             input = ChatMessageCreate(
                 id=str(XID()),
                 agent_id=cached_bot_item.agent_id,
                 chat_id=pool.agent_chat_id(
                     cached_bot_item.is_public_memory, message.chat.id
                 ),
-                user_id=cached_bot_item.agent_owner
-                if is_owner
-                else str(message.from_user.id),
-                author_id=str(message.from_user.id),
+                user_id=cached_bot_item.agent_owner if is_owner else user_id,
+                author_id=user_id,
                 author_type=AuthorType.TELEGRAM,
                 thread_type=AuthorType.TELEGRAM,
-                message=message_text,
+                message=wrapped_message,
             )
             response = await execute_agent(input)
             await message.answer(
@@ -122,6 +288,10 @@ async def gp_process_message(message: Message) -> None:
                 parse_mode="MarkdownV2",
                 reply_to_message_id=message.message_id,
             )
+
+            # Clear cached context after successful agent response
+            await clear_cached_context(message.chat.id, cached_bot_item.agent_id)
+
         except Exception as e:
             logger.warning(
                 f"error processing in function:{cur_func_name()}, token:{message.bot.token}, err={str(e)}"
@@ -135,6 +305,11 @@ async def gp_process_message(message: Message) -> None:
                 await message.react([])
             except Exception as e:
                 logger.warning(f"Failed to remove reaction: {str(e)}")
+    else:
+        # This message doesn't trigger a reply, cache it for context
+        cached_bot_item = pool.bot_by_token(message.bot.token)
+        if cached_bot_item:
+            await cache_message(message.chat.id, cached_bot_item.agent_id, message)
 
 
 ## direct commands and messages
@@ -164,11 +339,20 @@ async def process_message(message: Message) -> None:
         logger.warning(f"bot with token {message.bot.token} not found in cache.")
         return
 
-    if message.from_user:
+    try:
+        user_id = await get_user_id(message.from_user)
+        is_owner = (
+            cached_bot_item._owner == message.from_user.username
+            if message.from_user
+            else False
+        )
         logger.info(f"message from: {message.from_user}")
-        is_owner = cached_bot_item._owner == message.from_user.username
-    else:
+    except ValueError:
         is_owner = False
+        logger.error(
+            f"telegram message from user without username: {message.from_user}"
+        )
+        return
 
     # Add processing reaction
     await message.react([ReactionTypeEmoji(emoji="🤔")])
@@ -181,7 +365,7 @@ async def process_message(message: Message) -> None:
                     f"length: {len(message.text)}\n"
                     f"chat_id:{message.chat.id}\n"
                     f"agent:{cached_bot_item.agent_id}\n"
-                    f"user:{message.from_user.id}\n"
+                    f"user:{user_id}\n"
                     f"content:{message.text[:100]}..."
                 )
             )
@@ -190,10 +374,8 @@ async def process_message(message: Message) -> None:
             id=str(XID()),
             agent_id=cached_bot_item.agent_id,
             chat_id=pool.agent_chat_id(False, message.chat.id),
-            user_id=cached_bot_item.agent_owner
-            if is_owner
-            else str(message.from_user.id),
-            author_id=str(message.from_user.id),
+            user_id=cached_bot_item.agent_owner if is_owner else user_id,
+            author_id=user_id,
             author_type=AuthorType.TELEGRAM,
             thread_type=AuthorType.TELEGRAM,
             message=message.text,
diff --git a/intentkit/core/engine.py b/intentkit/core/engine.py
@@ -293,6 +293,12 @@ async def stream_agent_raw(
     # save input message first
     user_message = await message.save()
 
+    # temporary debug logging for telegram messages
+    if user_message.author_type == AuthorType.TELEGRAM:
+        logger.info(
+            f"[TELEGRAM DEBUG] Agent: {user_message.agent_id} | Chat: {user_message.chat_id} | Message: {user_message.message}"
+        )
+
     if re.search(
         r"(@clear|/clear)(?!\w)",
         user_message.message.strip(),
diff --git a/intentkit/core/prompt.py b/intentkit/core/prompt.py
@@ -12,6 +12,7 @@
 from intentkit.models.agent_data import AgentData
 from intentkit.models.chat import AuthorType
 from intentkit.models.skill import Skill
+from intentkit.models.user import User
 
 # ============================================================================
 # CONSTANTS AND CONFIGURATION
@@ -120,18 +121,37 @@ def _build_wallet_section(agent: Agent, agent_data: AgentData) -> str:
     return "\n".join(wallet_parts) + ("\n" if wallet_parts else "")
 
 
-def _build_user_info_section(context: AgentContext) -> str:
+async def _build_user_info_section(context: AgentContext) -> str:
     """Build user information section when user_id is a valid EVM wallet address."""
     if not context.user_id:
         return ""
 
-    # Check if user_id is a valid EVM wallet address
-    try:
-        if is_address(context.user_id):
-            return f"## User Info\n\nThe person you are talking to has wallet address: {context.user_id}\n\n"
-    except Exception:
-        # If validation fails, don't include the section
-        pass
+    user = await User.get(context.user_id)
+
+    prompt_array = []
+
+    evm_wallet_address = ""
+    if user and user.evm_wallet_address:
+        evm_wallet_address = user.evm_wallet_address
+    elif is_address(context.user_id):
+        evm_wallet_address = context.user_id
+
+    if evm_wallet_address:
+        prompt_array.append(
+            f"The user you are talking to has EVM wallet address: {evm_wallet_address}\n"
+        )
+
+    if user:
+        if user.email:
+            prompt_array.append(f"User Email: {user.email}\n")
+        if user.x_username:
+            prompt_array.append(f"User X Username: {user.x_username}\n")
+        if user.telegram_username:
+            prompt_array.append(f"User Telegram Username: {user.telegram_username}\n")
+
+    if prompt_array:
+        prompt_array.append("\n")
+        return "## User Info\n\n" + "".join(prompt_array)
 
     return ""
 
@@ -397,7 +417,7 @@ async def formatted_prompt(
             )
 
         # Add user info if user_id is a valid EVM wallet address
-        user_info = _build_user_info_section(context)
+        user_info = await _build_user_info_section(context)
         if user_info:
             final_system_prompt = f"{final_system_prompt}{user_info}"
 
diff --git a/intentkit/models/user.py b/intentkit/models/user.py
