Upload 6 files

aworld/core/context/processor/base_compressor.py

@@ -0,0 +1,24 @@
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+from aworld.config.conf import ModelConfig
+from aworld.core.context.processor import CompressionResult
+
+
+class BaseCompressor(ABC):
+    """Base compressor class"""
+    
+    def __init__(self, config: Dict[str, Any] = None, llm_config: ModelConfig = None):
+        self.config = config or {}
+        self.llm_config = llm_config
+    @abstractmethod
+    def compress(self, content: str, metadata: Dict[str, Any] = None) -> CompressionResult:
+        """Compress content"""
+        pass
+    
+    def _calculate_compression_ratio(self, original: str, compressed: str) -> float:
+        """Calculate compression ratio"""
+        if len(original) == 0:
+            return 1.0
+        return len(compressed) / len(original)

aworld/core/context/processor/chunk_utils.py

@@ -0,0 +1,433 @@
+import logging
+import time
+from dataclasses import dataclass
+from enum import Enum
+from typing import List, Dict, Any, Union
+
+from aworld.core.context.processor import MessageChunk, ChunkResult, MessageType
+
+logger = logging.getLogger(__name__)
+
+
+class ChunkUtils:
+    
+    def __init__(self, 
+                 enable_chunking: bool = False,
+                 preserve_order: bool = True,
+                 merge_consecutive: bool = True,
+                 max_chunk_size: int = None,
+                 split_by_tool_name: bool = False):
+        
+        # Chunker configuration
+        self.enable_chunking = enable_chunking
+        self.preserve_order = preserve_order
+        self.merge_consecutive = merge_consecutive
+        self.max_chunk_size = max_chunk_size
+        self.split_by_tool_name = split_by_tool_name
+        
+        # Statistics
+        self.stats = {
+            # Chunking statistics
+            "chunking": {
+                "total_processed": 0,
+                "total_chunks_created": 0,
+                "processing_time": 0.0
+            }
+        }
+
+    def _process_chunking(self, messages: List[Dict[str, Any]], **kwargs) -> List[Dict[str, Any]]:
+        """Process chunking logic"""
+        # First chunk
+        chunk_result = self.split_messages(messages, kwargs.get('metadata', {}))
+        
+        # Then merge back to message list
+        merged_messages = self.merge_chunks(chunk_result.chunks, 
+                                          kwargs.get('preserve_type_order', True))
+        
+        return merged_messages
+
+    def classify_message(self, message: Dict[str, Any]) -> MessageType:
+        """
+        Classify a single message
+        
+        Args:
+            message: OpenAI format message
+            
+        Returns:
+            Message type
+        """
+        role = message.get("role", "")
+        
+        if role in ["system", "user", "assistant"]:
+            return MessageType.TEXT
+        elif role == "tool":
+            return MessageType.TOOL
+        else:
+            logger.warning(f"Unknown message role: {role}")
+            return MessageType.UNKNOWN
+
+    def split_messages(self, 
+                      messages: List[Dict[str, Any]], 
+                      metadata: Dict[str, Any] = None) -> ChunkResult:
+        """
+        Split message list into chunks by type, and merge messages of the same type into strings
+        
+        Args:
+            messages: OpenAI format message list
+            metadata: Metadata
+            
+        Returns:
+            Chunking result
+        """
+        start_time = time.time()
+        
+        if not messages:
+            return ChunkResult(
+                chunks=[],
+                total_messages=0,
+                processing_time=0.0,
+                metadata=metadata or {}
+            )
+        
+        chunks = []
+        current_chunk_type = None
+        current_chunk_messages = []
+        
+        for i, message in enumerate(messages):
+            msg_type = self.classify_message(message)
+            
+            # If it's a new type or not merging consecutive messages
+            if (current_chunk_type != msg_type or 
+                not self.merge_consecutive):
+                
+                # Save current chunk (if has content)
+                if current_chunk_messages:
+                    chunk_metadata = (metadata or {}).copy()
+                    chunk_metadata.update({
+                        "chunk_index": len(chunks),
+                        "start_message_index": i - len(current_chunk_messages),
+                        "end_message_index": i - 1,
+                        "message_count": len(current_chunk_messages),
+                        "original_messages": current_chunk_messages.copy()
+                    })
+                    
+                    # Merge messages into strings based on message type
+                    if current_chunk_type == MessageType.TEXT:
+                        merged_content = self._messages_to_string(current_chunk_messages)
+                        merged_message = {
+                            "role": "merged_text",
+                            "content": merged_content,
+                            "original_count": len(current_chunk_messages)
+                        }
+                        chunk_messages = [merged_message]
+                    elif current_chunk_type == MessageType.TOOL:
+                        merged_content = self._tool_messages_to_string(current_chunk_messages)
+                        merged_message = {
+                            "role": "merged_tool",
+                            "content": merged_content,
+                            "original_count": len(current_chunk_messages)
+                        }
+                        chunk_messages = [merged_message]
+                    else:
+                        # Unknown type keeps as is
+                        chunk_messages = current_chunk_messages.copy()
+                    
+                    chunks.append(MessageChunk(
+                        message_type=current_chunk_type,
+                        messages=chunk_messages,
+                        metadata=chunk_metadata
+                    ))
+                
+                # Start new chunk
+                current_chunk_type = msg_type
+                current_chunk_messages = [message]
+            else:
+                # Add to current chunk
+                current_chunk_messages.append(message)
+        
+        # Process the last chunk
+        if current_chunk_messages:
+            chunk_metadata = (metadata or {}).copy()
+            chunk_metadata.update({
+                "chunk_index": len(chunks),
+                "start_message_index": len(messages) - len(current_chunk_messages),
+                "end_message_index": len(messages) - 1,
+                "message_count": len(current_chunk_messages),
+                "original_messages": current_chunk_messages.copy()
+            })
+            
+            # Merge messages into strings based on message type
+            if current_chunk_type == MessageType.TEXT:
+                merged_content = self._messages_to_string(current_chunk_messages)
+                merged_message = {
+                    "role": "merged_text",
+                    "content": merged_content,
+                    "original_count": len(current_chunk_messages)
+                }
+                chunk_messages = [merged_message]
+            elif current_chunk_type == MessageType.TOOL:
+                merged_content = self._tool_messages_to_string(current_chunk_messages)
+                merged_message = {
+                    "role": "merged_tool",
+                    "content": merged_content,
+                    "original_count": len(current_chunk_messages)
+                }
+                chunk_messages = [merged_message]
+            else:
+                chunk_messages = current_chunk_messages.copy()
+            
+            chunks.append(MessageChunk(
+                message_type=current_chunk_type,
+                messages=chunk_messages,
+                metadata=chunk_metadata
+            ))
+        
+        processing_time = time.time() - start_time
+        
+        # Update statistics
+        self.stats["chunking"]["total_processed"] += len(messages)
+        self.stats["chunking"]["total_chunks_created"] += len(chunks)
+        self.stats["chunking"]["processing_time"] += processing_time
+        
+        # Build result metadata
+        result_metadata = (metadata or {}).copy()
+        result_metadata.update({
+            "chunk_count": len(chunks),
+            "text_chunks": sum(1 for chunk in chunks if chunk.message_type == MessageType.TEXT),
+            "tool_chunks": sum(1 for chunk in chunks if chunk.message_type == MessageType.TOOL),
+            "unknown_chunks": sum(1 for chunk in chunks if chunk.message_type == MessageType.UNKNOWN),
+            "preserve_order": self.preserve_order,
+            "merge_consecutive": self.merge_consecutive,
+            "processing_time": processing_time,
+            "string_merge_applied": True
+        })
+        
+        logger.debug(f"Message splitting completed: {len(messages)} messages -> {len(chunks)} chunks (string merge applied)")
+        
+        return ChunkResult(
+            chunks=chunks,
+            total_messages=len(messages),
+            processing_time=processing_time,
+            metadata=result_metadata
+        )
+
+    def merge_chunks(self, 
+                    chunks: List[MessageChunk], 
+                    preserve_type_order: bool = True) -> List[Dict[str, Any]]:
+        """
+        Merge processed chunks back to message list, and split string format messages back to multiple messages
+        
+        Args:
+            chunks: Message chunk list
+            preserve_type_order: Whether to preserve type order
+            
+        Returns:
+            Merged message list
+        """
+        if not chunks:
+            return []
+        
+        if preserve_type_order and self.preserve_order:
+            # Merge in original order
+            sorted_chunks = sorted(chunks, key=lambda x: x.metadata.get("chunk_index", 0))
+        else:
+            # Merge by type groups (text first, then tools)
+            text_chunks = [chunk for chunk in chunks if chunk.message_type == MessageType.TEXT]
+            tool_chunks = [chunk for chunk in chunks if chunk.message_type == MessageType.TOOL]
+            unknown_chunks = [chunk for chunk in chunks if chunk.message_type == MessageType.UNKNOWN]
+            sorted_chunks = text_chunks + tool_chunks + unknown_chunks
+        
+        merged_messages = []
+        for chunk in sorted_chunks:
+            chunk_messages = []
+            
+            for message in chunk.messages:
+                # Check if it's a merged message that needs splitting
+                if message.get("role") == "merged_text":
+                    # This is a merged TEXT type message that needs splitting
+                    merged_content = message.get("content", "")
+                    original_messages = chunk.metadata.get("original_messages", [])
+                    
+                    if original_messages:
+                        split_messages = self._string_to_messages(merged_content, original_messages)
+                        chunk_messages.extend(split_messages)
+                    else:
+                        split_messages = self._string_to_messages(merged_content, [])
+                        chunk_messages.extend(split_messages)
+                        
+                elif message.get("role") == "merged_tool":
+                    # This is a merged TOOL type message that needs splitting
+                    merged_content = message.get("content", "")
+                    original_messages = chunk.metadata.get("original_messages", [])
+                    
+                    if original_messages:
+                        split_messages = self._string_to_tool_messages(merged_content, original_messages)
+                        chunk_messages.extend(split_messages)
+                    else:
+                        split_messages = self._string_to_tool_messages(merged_content, "")
+                        chunk_messages.extend(split_messages)
+                        
+                else:
+                    # Regular message added directly
+                    chunk_messages.append(message)
+            
+            merged_messages.extend(chunk_messages)
+        
+        return merged_messages
+
+    # Message conversion methods
+    @staticmethod
+    def _messages_to_string(messages: List[Dict[str, str]]) -> str:
+        """Convert OpenAI message format to string"""
+        content_parts = []
+        for msg in messages:
+            role = msg.get('role', 'user')
+            content = msg.get('content', '')
+            content_parts.append(f"[{role.upper()}]: {content}")
+        return "\n".join(content_parts)
+    
+    @staticmethod
+    def _string_to_messages(content: str, messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
+        """Convert string to OpenAI message format"""
+        # Restore all tool_calls
+        tool_calls = []
+        if messages:
+            for msg in messages:
+                if msg.get("role") == "assistant" and msg.get("tool_calls") is not None:
+                    tool_calls += msg["tool_calls"]
+
+        result_messages = []
+        lines = content.split('\n')
+        current_role = 'user'
+        current_content = []
+        
+        for line in lines:
+            line = line.strip()
+            if line.startswith('[') and ']:' in line:
+                # Save previous message
+                if current_content:
+                    result_messages.append({
+                        'role': current_role,
+                        'content': '\n'.join(current_content).strip()
+                    })
+                    current_content = []
+                
+                # Parse new role
+                role_end = line.find(']:')
+                role = line[1:role_end].lower()
+                if role in ['system', 'user', 'assistant']:
+                    current_role = role
+                    content_part = line[role_end + 2:].strip()
+                    if content_part:
+                        current_content.append(content_part)
+                else:
+                    current_content.append(line)
+            else:
+                current_content.append(line)
+
+        # Save last message
+        if current_content:
+            result_messages.append({
+                'role': current_role,
+                'content': '\n'.join(current_content).strip(),
+            })
+        
+        final_messages = result_messages if result_messages else [{'role': 'user', 'content': content}]
+
+        # Add tool_calls results
+        if tool_calls and len(tool_calls) > 0:
+            tool_call_chunk = {
+                'role': 'assistant',
+                'content': None,
+                'tool_calls': tool_calls
+            }
+            final_messages.append(tool_call_chunk)
+
+        return final_messages
+
+    def _tool_messages_to_string(self, messages: List[Dict[str, str]]) -> str:
+        """Convert tool message format to string"""
+        content_parts = []
+        for msg in messages:
+            role = msg.get('role', 'tool')
+            content = msg.get('content', '')
+            tool_call_id = msg.get('tool_call_id', '')
+            name = msg.get('name', '')
+    
+            if role == 'tool':
+                header = f"[TOOL:{name}:{tool_call_id}]"
+            else:
+                header = f"[{role.upper()}]"
+    
+            content_parts.append(f"{header}: {content}")
+        return "\n".join(content_parts)
+    
+    def _string_to_tool_messages(self, content: str, original_prompt: Union[str, List[Dict[str, str]]]) -> List[Dict[str, str]]:
+        """Convert string to tool message format"""
+        messages = []
+        lines = content.split('\n')
+        current_role = 'tool'
+        current_content = []
+        current_tool_call_id = ''
+        current_name = ''
+    
+        for line in lines:
+            line = line.strip()
+            if line.startswith('[') and ']:' in line:
+                # Save previous message
+                if current_content:
+                    msg = {
+                        'role': current_role,
+                        'content': '\n'.join(current_content).strip()
+                    }
+                    if current_role == 'tool':
+                        if current_tool_call_id:
+                            msg['tool_call_id'] = current_tool_call_id
+                        if current_name:
+                            msg['name'] = current_name
+                    messages.append(msg)
+                    current_content = []
+    
+                # Parse new role and tool information
+                role_end = line.find(']:')
+                role_part = line[1:role_end]
+                content_part = line[role_end + 2:].strip()
+    
+                if role_part.startswith('TOOL:'):
+                    # Parse tool message format: [TOOL:name:tool_call_id]
+                    current_role = 'tool'
+                    tool_parts = role_part.split(':')
+                    if len(tool_parts) >= 2:
+                        current_name = tool_parts[1]
+                    if len(tool_parts) >= 3:
+                        current_tool_call_id = tool_parts[2]
+                else:
+                    current_role = role_part.lower()
+                    current_tool_call_id = ''
+                    current_name = ''
+    
+                if content_part:
+                    current_content.append(content_part)
+            else:
+                current_content.append(line)
+    
+        # Save last message
+        if current_content:
+            msg = {
+                'role': current_role,
+                'content': '\n'.join(current_content).strip()
+            }
+            if current_role == 'tool':
+                if current_tool_call_id:
+                    msg['tool_call_id'] = current_tool_call_id
+                if current_name:
+                    msg['name'] = current_name
+            messages.append(msg)
+    
+        # If no messages parsed, return original format
+        if not messages and isinstance(original_prompt, list):
+            return original_prompt
+        elif not messages:
+            return [{'role': 'tool', 'content': content}]
+    
+        return messages

aworld/core/context/processor/llm_compressor.py

@@ -0,0 +1,113 @@
+import asyncio
+import logging
+import re
+from abc import ABC, abstractmethod
+import traceback
+from typing import Any, Dict, List
+
+from aworld.config.conf import ModelConfig
+from aworld.core.context.processor import CompressionResult, CompressionType
+from aworld.core.context.processor.base_compressor import BaseCompressor
+from aworld.models.llm import get_llm_model
+from aworld.config import ConfigDict
+                
+logger = logging.getLogger(__name__)
+
+class LLMCompressor(BaseCompressor):
+    """LLM-based prompt compressor"""
+    
+    def __init__(self, config: Dict[str, Any] = None, llm_config: ModelConfig = None):
+        super().__init__(config, llm_config)
+        self.compression_prompt = self.config.get("compression_prompt", self._default_compression_prompt())
+        # Lazy import to avoid circular dependencies
+        self._llm_client = self._create_llm_client(llm_config)
+    
+    @staticmethod
+    def _remove_think_blocks(content: str) -> str:
+        """Remove <think>...</think> blocks from content"""
+        # Use regex to remove all <think>...</think> blocks (case insensitive, multiline)
+        pattern = r'<think>.*?</think>'
+        cleaned_content = re.sub(pattern, '', content, flags=re.IGNORECASE | re.DOTALL)
+        return cleaned_content
+
+    def _create_llm_client(self, llm_config: ModelConfig):
+        if llm_config is None:
+            return None
+        config = ConfigDict(llm_config.model_dump())
+        return get_llm_model(config)
+
+    def _default_compression_prompt(self) -> str:
+        """Default compression prompt"""
+        return """## Task
+You are a text compression expert. Please intelligently compress the following text, retaining core information and key content while removing redundancy and unimportant parts.
+
+## Compression Requirements
+1. Keep the position and count of [SYSTEM], [USER], [ASSISTANT], and [TOOL] tags unchanged in the output
+2. Maintain the main meaning and logical structure of the original text, retain key information and important details, use more concise expressions
+3. Remove repetitive, redundant statements, ensure the compressed text remains coherent and readable
+
+## Original Text:
+{content}
+
+Please output the compressed text:"""
+    
+    def compress(self, content: str) -> CompressionResult:
+        """Compress content using LLM"""
+        original_content = content
+        
+        # Get LLM client
+        llm_client = self._llm_client
+        if llm_client is None:
+            logger.warning("LLM client unavailable, returning original content")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=content,
+                compression_ratio=1.0,
+                metadata={"error": "LLM client unavailable"},
+                compression_type=CompressionType.LLM_BASED
+            )
+        
+        try:
+            # Build prompt
+            prompt = self.compression_prompt.format(content=content)
+            messages = [{"role": "user", "content": prompt}]
+            
+            # Call LLM
+            response = llm_client.completion(
+                messages=messages,
+                temperature=0.3
+            )
+            
+            # Remove <think>...</think> blocks first, then strip whitespace
+            compressed_content = self._remove_think_blocks(response.content).strip()
+            compression_ratio = self._calculate_compression_ratio(original_content, compressed_content)
+            
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=compressed_content,
+                compression_ratio=compression_ratio,
+                metadata={
+                    "prompt_tokens": getattr(response.usage, 'prompt_tokens', 0),
+                    "completion_tokens": getattr(response.usage, 'completion_tokens', 0),
+                },
+                compression_type=CompressionType.LLM_BASED
+            )
+            
+        except Exception as e:
+            logger.error(f"LLM compression failed: {traceback.format_exc()}")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=content,
+                compression_ratio=1.0,
+                metadata={"error": str(e)},
+                compression_type=CompressionType.LLM_BASED
+            )
+    
+    def compress_batch(self, contents: List[str]) -> List[CompressionResult]:
+        """Compress multiple contents in batch"""
+        results = []
+        for content in contents:
+            result = self.compress(content)
+            results.append(result)
+        return results
+    

aworld/core/context/processor/llmlingua_compressor.py

@@ -0,0 +1,295 @@
+import re
+import logging
+from typing import Any, Dict, List, Optional, Pattern, Tuple
+
+from aworld.config.conf import ModelConfig
+from aworld.core.context.processor import CompressionResult, CompressionType
+from aworld.core.context.processor.base_compressor import BaseCompressor
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_LLM_LINGUA_INSTRUCTION = (
+    "Given this conversation messages, please compress them while preserving key information"
+)
+
+
+class LLMLinguaCompressor(BaseCompressor):
+    """
+    Compress messages using LLMLingua Project.
+    
+    https://github.com/microsoft/LLMLingua
+    """
+
+    # Pattern to match ref tags at the beginning or end of the string,
+    # allowing for malformed tags
+    _pattern_beginning: Pattern = re.compile(r"\A(?:<#)?(?:ref)?(\d+)(?:#>?)?")
+    _pattern_ending: Pattern = re.compile(r"(?:<#)?(?:ref)?(\d+)(?:#>?)?\Z")
+
+    def __init__(self, config: Dict[str, Any] = None, llm_config: ModelConfig = None):
+        super().__init__(config, llm_config)
+        
+        # LLMLingua specific configuration
+        self.model_name = self.config.get("model_name", "NousResearch/Llama-2-7b-hf")
+        self.device_map = self.config.get("device_map", "cuda")
+        self.target_token = self.config.get("target_token", 300)
+        self.rank_method = self.config.get("rank_method", "longllmlingua")
+        self.model_configuration = self.config.get("model_configuration", {})
+        self.open_api_config = self.config.get("open_api_config", {})
+        self.instruction = self.config.get("instruction", DEFAULT_LLM_LINGUA_INSTRUCTION)
+        self.additional_compress_kwargs = self.config.get("additional_compress_kwargs", {
+            "condition_compare": True,
+            "condition_in_question": "after",
+            "context_budget": "+100",
+            "reorder_context": "sort",
+            "dynamic_context_compression_ratio": 0.4,
+        })
+        
+        self.lingua = None
+        self._initialize_lingua()
+
+    def _initialize_lingua(self):
+        """Initialize LLMLingua PromptCompressor"""
+        try:
+            from llmlingua import PromptCompressor
+            
+            self.lingua = PromptCompressor(
+                model_name=self.model_name,
+                device_map=self.device_map,
+                model_config=self.model_configuration,
+                open_api_config=self.open_api_config,
+            )
+            logger.info(f"LLMLingua compressor initialized with model: {self.model_name}")
+            
+        except ImportError:
+            logger.error(
+                "Could not import llmlingua python package. "
+                "Please install it with `pip install llmlingua`."
+            )
+            self.lingua = None
+        except Exception as e:
+            logger.error(f"Failed to initialize LLMLingua compressor: {e}")
+            self.lingua = None
+
+    @staticmethod
+    def _format_messages(messages: List[Dict[str, Any]]) -> List[str]:
+        """
+        Format messages by including special ref tags for tracking after compression
+        """
+        formatted_messages = []
+        for i, message in enumerate(messages):
+            role = message.get("role", "unknown")
+            content = message.get("content", "").replace("\n\n", "\n")
+            
+            # Format as [ROLE] content with ref tags
+            message_string = f"\n\n<#ref{i}#> [{role.upper()}] {content} <#ref{i}#>\n\n"
+            formatted_messages.append(message_string)
+        return formatted_messages
+
+    def extract_ref_id_tuples_and_clean(self, contents: List[str]) -> List[Tuple[str, int]]:
+        """
+        Extracts reference IDs from the contents and cleans up the ref tags.
+        
+        Args:
+            contents: A list of contents to be processed.
+
+        Returns:
+            List of tuples containing (cleaned_string, ref_id)
+        """
+        ref_id_tuples = []
+        for content in contents:
+            clean_string = content.strip()
+            if not clean_string:
+                continue
+
+            # Search for ref tags at the beginning and the end of the string
+            ref_id = None
+            for pattern in [self._pattern_beginning, self._pattern_ending]:
+                match = pattern.search(clean_string)
+                if match:
+                    ref_id = match.group(1)
+                    clean_string = pattern.sub("", clean_string).strip()
+            
+            # Convert ref ID to int or use -1 if not found
+            ref_id_to_use = int(ref_id) if ref_id and ref_id.isdigit() else -1
+            ref_id_tuples.append((clean_string, ref_id_to_use))
+
+        return ref_id_tuples
+
+    def _parse_compressed_content_to_messages(self, compressed_content: str, original_messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        Parse compressed content back to message format
+        """
+        # Split by double newlines and filter empty strings
+        compressed_parts = [part.strip() for part in compressed_content.split("\n\n") if part.strip()]
+        
+        extracted_metadata = self.extract_ref_id_tuples_and_clean(compressed_parts)
+        
+        compressed_messages = []
+        for content, index in extracted_metadata:
+            if not content:
+                continue
+                
+            # Parse role from content if present
+            role_match = re.match(r'\[(\w+)\]\s*(.*)', content)
+            if role_match:
+                role = role_match.group(1).lower()
+                message_content = role_match.group(2).strip()
+            else:
+                # Fallback to original message role if available
+                role = "assistant"  # Default role
+                message_content = content
+                if index != -1 and index < len(original_messages):
+                    role = original_messages[index].get("role", "assistant")
+            
+            compressed_messages.append({
+                "role": role,
+                "content": message_content
+            })
+        
+        return compressed_messages
+
+    def compress(self, content: str) -> CompressionResult:
+        """
+        Compress content using LLMLingua
+        
+        Note: This method expects content to be a JSON string representation of messages
+        or will treat it as a single message.
+        """
+        original_content = content
+        
+        if self.lingua is None:
+            logger.warning("LLMLingua compressor unavailable, returning original content")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=content,
+                compression_ratio=1.0,
+                metadata={"error": "LLMLingua compressor unavailable"},
+                compression_type=CompressionType.LLMLINGUA
+            )
+        
+        try:
+            # Try to parse as messages format first
+            import json
+            try:
+                messages = json.loads(content)
+                if isinstance(messages, list) and all(isinstance(msg, dict) for msg in messages):
+                    return self.compress_messages(messages)
+            except (json.JSONDecodeError, TypeError):
+                pass
+            
+            # Treat as plain text
+            formatted_content = [f"\n\n<#ref0#> {content} <#ref0#>\n\n"]
+            
+            compressed_prompt = self.lingua.compress_prompt(
+                context=formatted_content,
+                instruction=self.instruction,
+                question="",  # No specific question for plain text
+                target_token=self.target_token,
+                rank_method=self.rank_method,
+                concate_question=False,
+                add_instruction=False,
+                **self.additional_compress_kwargs,
+            )
+            
+            compressed_content = compressed_prompt["compressed_prompt"]
+            compression_ratio = self._calculate_compression_ratio(original_content, compressed_content)
+            
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=compressed_content,
+                compression_ratio=compression_ratio,
+                metadata={
+                    "origin_tokens": compressed_prompt.get("origin_tokens", 0),
+                    "compressed_tokens": compressed_prompt.get("compressed_tokens", 0),
+                    "ratio": compressed_prompt.get("ratio", "unknown"),
+                },
+                compression_type=CompressionType.LLMLINGUA
+            )
+            
+        except Exception as e:
+            logger.error(f"LLMLingua compression failed: {e}")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=content,
+                compression_ratio=1.0,
+                metadata={"error": str(e)},
+                compression_type=CompressionType.LLMLINGUA
+            )
+
+    def compress_messages(self, messages: List[Dict[str, Any]]) -> CompressionResult:
+        """
+        Compress a list of messages using LLMLingua
+        """
+        if not messages:
+            return CompressionResult(
+                original_content="[]",
+                compressed_content="[]",
+                compression_ratio=1.0,
+                metadata={},
+                compression_type=CompressionType.LLMLINGUA
+            )
+        
+        original_content = str(messages)
+        
+        if self.lingua is None:
+            logger.warning("LLMLingua compressor unavailable, returning original messages")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=original_content,
+                compression_ratio=1.0,
+                metadata={"error": "LLMLingua compressor unavailable"},
+                compression_type=CompressionType.LLMLINGUA
+            )
+        
+        try:
+            formatted_messages = self._format_messages(messages)
+            
+            compressed_prompt = self.lingua.compress_prompt(
+                context=formatted_messages,
+                instruction=self.instruction,
+                question="",  # No specific question for conversation compression
+                target_token=self.target_token,
+                rank_method=self.rank_method,
+                concate_question=False,
+                add_instruction=False,
+                **self.additional_compress_kwargs,
+            )
+            
+            # Parse compressed content back to messages
+            compressed_messages = self._parse_compressed_content_to_messages(
+                compressed_prompt["compressed_prompt"], messages
+            )
+            
+            compressed_content = str(compressed_messages)
+            compression_ratio = self._calculate_compression_ratio(original_content, compressed_content)
+            
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=compressed_content,
+                compression_ratio=compression_ratio,
+                metadata={
+                    "origin_tokens": compressed_prompt.get("origin_tokens", 0),
+                    "compressed_tokens": compressed_prompt.get("compressed_tokens", 0),
+                    "ratio": compressed_prompt.get("ratio", "unknown"),
+                    "compressed_messages": compressed_messages,
+                },
+                compression_type=CompressionType.LLMLINGUA
+            )
+            
+        except Exception as e:
+            logger.error(f"LLMLingua message compression failed: {e}")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=original_content,
+                compression_ratio=1.0,
+                metadata={"error": str(e)},
+                compression_type=CompressionType.LLMLINGUA
+            )
+
+    def compress_batch(self, contents: List[str]) -> List[CompressionResult]:
+        """Compress multiple contents in batch"""
+        results = []
+        for content in contents:
+            result = self.compress(content)
+            results.append(result)
+        return results

aworld/core/context/processor/prompt_processor.py

@@ -0,0 +1,455 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+import time
+from dataclasses import dataclass
+import traceback
+from typing import Dict, Any, List
+
+from aworld.core.context.base import Context, AgentContext
+from aworld.core.context.processor import CompressionDecision, ContextProcessingResult, MessagesProcessingResult
+from aworld.core.context.processor.llm_compressor import LLMCompressor, CompressionType
+from aworld.core.context.processor.llmlingua_compressor import LLMLinguaCompressor
+from aworld.core.context.processor.truncate_compressor import TruncateCompressor
+from aworld.core.context.processor.chunk_utils import ChunkUtils, MessageChunk, MessageType
+from aworld.logs.util import Color, color_log, logger
+from aworld.models.utils import num_tokens_from_messages, truncate_tokens_from_messages
+from aworld.config.conf import AgentConfig, ConfigDict, ContextRuleConfig, ModelConfig, OptimizationConfig, LlmCompressionConfig
+
+class PromptProcessor:
+    """Agent context processor, processes context according to context_rule configuration"""
+    
+    def __init__(self, agent_context: AgentContext):
+        self.context_rule = agent_context.context_rule
+        self.agent_context = agent_context
+        self.compress_pipeline = None
+        self.llmlingua_compressor = None
+        self.truncate_compressor = None
+        self.chunk_pipeline = None
+        self._init_pipelines()
+    
+    def _init_pipelines(self): 
+        """Initialize processing pipelines"""
+        # Initialize truncate compressor
+        self.truncate_compressor = TruncateCompressor(
+            config={},
+            llm_config=self.agent_context.model_config
+        )
+        
+        if self.context_rule and self.context_rule.llm_compression_config and self.context_rule.llm_compression_config.enabled:
+            # Initialize message splitting and compression pipeline
+            self.chunk_pipeline = ChunkUtils(
+                enable_chunking=True,
+                preserve_order=True,
+                merge_consecutive=True,
+            )
+            
+            # Initialize compression pipeline based on compress_type configuration
+            compress_type = self.context_rule.llm_compression_config.compress_type
+            
+            if compress_type == 'llmlingua':
+                # Initialize LLMLingua compressor
+                self.llmlingua_compressor = LLMLinguaCompressor(
+                    config=getattr(self.context_rule.llm_compression_config, 'llmlingua_config', {}),
+                    llm_config=self.agent_context.context_rule.llm_compression_config.compress_model,
+                )
+            else:
+                # Default to LLM-based compression
+                self.compress_pipeline = LLMCompressor(
+                    config=getattr(self.context_rule.llm_compression_config, 'llm_config', {}),
+                    llm_config=self.agent_context.context_rule.llm_compression_config.compress_model,
+                )
+    
+    def _get_compression_type(self) -> CompressionType:
+        """Get the compression type based on configuration"""
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return CompressionType.LLM_BASED
+        
+        compress_type = self.context_rule.llm_compression_config.compress_type
+        if compress_type == 'llmlingua':
+            return CompressionType.LLMLINGUA
+        else:
+            return CompressionType.LLM_BASED
+
+    def get_max_tokens(self):
+        return self.agent_context.context_usage.total_context_length * self.context_rule.optimization_config.max_token_budget_ratio
+
+    def is_out_of_context(self, messages: List[Dict[str, Any]],
+                          is_last_message_in_memory: bool) -> bool:
+        return self._count_tokens_from_messages(messages) > self.get_max_tokens()
+        # Calculate based on historical message length to determine if threshold is reached, this is a rough statistic
+        # current_usage = self.agent_context.context_usage
+        # real_used = current_usage.used_context_length
+        # if not is_last_message_in_memory:
+        #     real_used += self._count_tokens_from_message(messages[-1])
+        # return real_used > self.get_max_tokens()
+
+    def _count_tokens_from_messages(self, messages: List[Dict[str, Any]]) -> int:
+        """Calculate token count for messages using utils.py method"""
+        return num_tokens_from_messages(messages, model=self.agent_context.model_config.model_type)
+
+    def _count_tokens_from_message(self, msg: Dict[str, Any]) -> int:
+        """Calculate token count for single message using utils.py method"""
+        # Convert single message to list format for num_tokens_from_messages
+        return num_tokens_from_messages([msg], model=self.agent_context.model_config.model_type)
+
+    def _count_chunk_tokens(self, chunk: MessageChunk) -> int:
+        """Calculate token count for a chunk"""
+        return num_tokens_from_messages(chunk.messages, model=self.agent_context.model_config.model_type)
+    
+    def _count_content_tokens(self, content: str) -> int:
+        """Calculate token count for content string"""
+        return num_tokens_from_messages(content, model=self.agent_context.model_config.model_type)
+
+    def _truncate_tokens_from_messages(self, content: str, max_tokens: int, keep_both_sides: bool = False) -> str:
+        """Calculate token count for messages using utils.py method"""
+        return truncate_tokens_from_messages(content, max_tokens, keep_both_sides, model=self.agent_context.model_config.model_type)
+
+    def decide_compression_strategy(self, chunk: MessageChunk) -> CompressionDecision:
+        """
+        Decide compression strategy based on chunk token length
+        
+        Args:
+            chunk: Message chunk to analyze
+            
+        Returns:
+            CompressionDecision with compression strategy
+        """
+        compression_type = self._get_compression_type()
+        
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return CompressionDecision(
+                should_compress=False,
+                compression_type=compression_type,
+                reason="Compression disabled in config",
+                token_count=0
+            )
+        
+        token_count = self._count_chunk_tokens(chunk)
+        trigger_compress_length = self.context_rule.llm_compression_config.trigger_compress_token_length
+        
+        # No compression needed
+        if token_count < trigger_compress_length:
+            return CompressionDecision(
+                should_compress=False,
+                compression_type=compression_type,
+                reason=f"Token count {token_count} below threshold {trigger_compress_length}",
+                token_count=token_count
+            )
+        
+        # Use configured compression for content above threshold
+        else:
+            return CompressionDecision(
+                should_compress=True,
+                compression_type=compression_type,
+                reason=f"Token count {token_count} exceeds threshold {trigger_compress_length}",
+                token_count=token_count
+            )
+
+    def decide_content_compression_strategy(self, content: str) -> CompressionDecision:
+        compression_type = self._get_compression_type()
+        
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return CompressionDecision(
+                should_compress=False,
+                compression_type=compression_type,
+                reason="Compression disabled in config",
+                token_count=0
+            )
+        
+        token_count = self._count_content_tokens(content)
+        trigger_compress_length = self.context_rule.llm_compression_config.trigger_compress_token_length
+        
+        # No compression needed
+        if token_count < trigger_compress_length:
+            return CompressionDecision(
+                should_compress=False,
+                compression_type=compression_type,
+                reason=f"Token count {token_count} below threshold {trigger_compress_length}",
+                token_count=token_count
+            )
+        
+        # Use configured compression for content above threshold
+        else:
+            return CompressionDecision(
+                should_compress=True,
+                compression_type=compression_type,
+                reason=f"Token count {token_count} exceeds threshold {trigger_compress_length}",
+                token_count=token_count
+            )
+
+    def should_compress_conversation(self, messages: List[Dict[str, Any]]) -> bool:
+        """Determine whether conversation compression is needed (legacy method for compatibility)"""
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return False
+        
+        # Create temporary chunk for decision
+        temp_chunk = MessageChunk(
+            message_type=MessageType.TEXT,
+            messages=messages,
+            metadata={}
+        )
+        
+        decision = self.decide_compression_strategy(temp_chunk)
+        return decision.should_compress
+    
+    def should_compress_tool_result(self, result: str) -> bool:
+        """Determine whether tool result compression is needed (legacy method for compatibility)"""
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return False
+        
+        decision = self.decide_content_compression_strategy(result)
+        return decision.should_compress
+    
+    def process_message_chunks(self, 
+                              chunks: List[MessageChunk], 
+                              base_metadata: Dict[str, Any] = None) -> List[MessageChunk]:
+        processed_chunks = []
+        
+        for chunk in chunks:
+            try:
+                if chunk.message_type == MessageType.TEXT:
+                    # Process text message chunks
+                    processed_chunk = self._process_text_chunk(chunk, base_metadata)
+                elif chunk.message_type == MessageType.TOOL:
+                    # Process tool message chunks
+                    processed_chunk = self._process_tool_chunk(chunk, base_metadata)
+                else:
+                    # Unknown type, keep as is
+                    processed_chunk = chunk
+                    logger.warning(f"Unknown message chunk type: {chunk.message_type}")
+                
+                processed_chunks.append(processed_chunk)
+                
+            except Exception as e:
+                logger.error(f"Processing message chunk failed: {traceback.format_exc()}")
+                # Keep original chunk on failure
+                processed_chunks.append(chunk)
+        
+        return processed_chunks
+    
+    def _process_text_chunk(self, 
+                           chunk: MessageChunk, 
+                           base_metadata: Dict[str, Any] = None) -> MessageChunk:
+        decision = self.decide_compression_strategy(chunk)
+        
+        if not decision.should_compress:
+            logger.debug(f"Skipping text chunk compression: {decision.reason}")
+            return chunk
+        
+        try:
+            processed_messages = []
+            
+            for message in chunk.messages:
+                content = message.get("content", "")
+                if not content or not isinstance(content, str):
+                    processed_messages.append(message)
+                    continue
+                
+                logger.info(f'Processing text chunk with LLM compression '
+                            f'(tokens: {decision.token_count}, reason: {decision.reason})')
+                
+                # Use LLM compression
+                compression_result = self.compress_pipeline.compress(content)
+                
+                # Create processed message
+                processed_message = message.copy()
+                processed_message["content"] = compression_result.compressed_content
+                processed_messages.append(processed_message)
+            
+            # Update chunk metadata
+            updated_metadata = chunk.metadata.copy()
+            updated_metadata.update({
+                "processed": True,
+                "compression_applied": True,
+                "compression_type": "llm_based",
+                "compression_reason": decision.reason,
+                "original_token_count": decision.token_count,
+                "processing_method": "llm_compression",
+                "original_message_count": len(chunk.messages),
+                "processed_message_count": len(processed_messages)
+            })
+            
+            return MessageChunk(
+                message_type=chunk.message_type,
+                messages=processed_messages,
+                metadata=updated_metadata
+            )
+            
+            return chunk
+            
+        except Exception as e:
+            logger.warning(f"Text chunk compression failed: {traceback.format_exc()}")
+            return chunk
+    
+    def _process_tool_chunk(self, 
+                           chunk: MessageChunk, 
+                           base_metadata: Dict[str, Any] = None) -> MessageChunk:
+        """Process tool message chunks with LLM compression"""
+        try:
+            processed_messages = []
+            
+            for message in chunk.messages:
+                content = message.get("content", "")
+                
+                # Decide compression strategy for this content
+                decision = self.decide_content_compression_strategy(content)
+                
+                if decision.should_compress:
+                    logger.info(f'Processing tool chunk with LLM compression '
+                              f'(tokens: {decision.token_count}, reason: {decision.reason})')
+                    
+                    # Use LLM compression
+                    compression_result = self.compress_pipeline.compress(
+                        content,
+                        metadata={
+                            "tool_name": message.get("name", "unknown_tool"),
+                            "message_role": message.get("role", "tool"),
+                            "content_token_count": decision.token_count,
+                            "compression_reason": decision.reason
+                        },
+                        compression_type=CompressionType.LLM_BASED
+                    )
+                    
+                    # Create processed message
+                    processed_message = message.copy()
+                    processed_message["content"] = compression_result.compressed_content
+                    processed_messages.append(processed_message)
+                else:
+                    # Messages that don't need compression are kept as is
+                    logger.debug(f"Skipping tool content compression: {decision.reason}")
+                    processed_messages.append(message)
+            
+            # Update chunk metadata with compression info
+            updated_metadata = chunk.metadata.copy()
+            updated_metadata.update({
+                "processed": True,
+                "tool_compression_applied": True,
+                "processing_method": "llm_compression",
+                "original_message_count": len(chunk.messages),
+                "processed_message_count": len(processed_messages)
+            })
+            
+            return MessageChunk(
+                message_type=chunk.message_type,
+                messages=processed_messages,
+                metadata=updated_metadata
+            )
+            
+        except Exception as e:
+            logger.warning(f"Tool chunk compression failed: {traceback.format_exc()}")
+            return chunk
+
+    def truncate_messages(self, messages: List[Dict[str, Any]]) -> MessagesProcessingResult:
+        """Truncate messages using TruncateCompressor"""
+        max_tokens = self.get_max_tokens()
+        optimization_enabled = self.context_rule.optimization_config.enabled if self.context_rule else True
+        
+        return self.truncate_compressor.truncate_messages(
+            messages=messages,
+            max_tokens=max_tokens,
+            optimization_enabled=optimization_enabled
+        )
+
+    def compress_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        if (not self.context_rule or 
+            not self.context_rule.llm_compression_config or 
+            not self.context_rule.llm_compression_config.enabled):
+            return messages
+        
+        compression_type = self._get_compression_type()
+        
+        if compression_type == CompressionType.LLMLINGUA and self.llmlingua_compressor:
+            # Use LLMLingua compression directly on messages
+            logger.info("Using LLMLingua compression for messages")
+            
+            try:
+                compression_result = self.llmlingua_compressor.compress_messages(messages)
+                
+                # Extract compressed messages from metadata
+                compressed_messages = compression_result.metadata.get("compressed_messages", messages)
+                
+                logger.info(f"LLMLingua compression completed. "
+                          f"Original: {len(messages)} messages, "
+                          f"Compressed: {len(compressed_messages)} messages, "
+                          f"Compression ratio: {compression_result.compression_ratio:.2f}")
+                
+                return compressed_messages
+                
+            except Exception as e:
+                logger.error(f"LLMLingua compression failed: {e}")
+                return messages
+        
+        elif compression_type == CompressionType.LLM_BASED and self.compress_pipeline:
+            # Use original chunk-based LLM compression
+            logger.info("Using LLM-based compression for messages")
+            
+            # 1. Re-split processed messages
+            final_chunk_result = self.chunk_pipeline.split_messages(messages)
+
+            # 2. Process each chunk
+            processed_chunks = self.process_message_chunks(final_chunk_result.chunks)
+            
+            # 3. Re-merge messages
+            return self.chunk_pipeline.merge_chunks(processed_chunks)
+        
+        else:
+            # No appropriate compressor available
+            logger.warning(f"No compressor available for type {compression_type}, returning original messages")
+            return messages
+
+    def process_messages(self, messages: List[Dict[str, Any]], context: Context) -> ContextProcessingResult:
+        """Process complete context, return processing results and statistics"""
+        start_time = time.time()
+        if not self.context_rule.optimization_config.enabled:
+            return ContextProcessingResult(
+                processed_messages=messages,
+                processed_tool_results=None,
+                statistics={
+                    "total_processing_time": 0,
+                    "original_message_count": len(messages),
+                },
+            )
+
+        # 1. Content compression
+        compressed_messages = self.compress_messages(messages)
+        
+        # 2. Content length limit
+        truncated_result = self.truncate_messages(compressed_messages)
+        truncated_messages = truncated_result.processed_messages
+        
+        total_time = time.time() - start_time
+
+        color_log(f"\nContext processing statistics: "
+                   f"\nOriginal message count={truncated_result.original_messages_len}"
+                   f"\nProcessed message count={truncated_result.processing_messaged_len}"
+                   f"\nMax context length max_context_len={self.get_max_tokens()} = {self.agent_context.context_usage.total_context_length} * {self.context_rule.optimization_config.max_token_budget_ratio}"
+                   f"\nOriginal token count={truncated_result.original_token_len}"
+                   f"\nProcessed token count={truncated_result.processing_token_len}"
+                   f"\nTruncation processing time={truncated_result.processing_time:.3f}s"
+                   f"\nTotal processing time={total_time:.3f}s"
+                   f"\nMethod used={truncated_result.method_used}"
+                   f"\norigin_messages={messages}"
+                   f"\ntruncated_messages={truncated_messages}",
+                   color=Color.pink,)
+
+        return ContextProcessingResult(
+            processed_messages=truncated_messages,
+            processed_tool_results=None,
+            statistics={
+                "total_processing_time": total_time,
+                "original_message_count": len(messages),
+                "truncated_message_count": len(truncated_messages),
+            },
+        ) 
+

aworld/core/context/processor/truncate_compressor.py

@@ -0,0 +1,355 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+import time
+import logging
+from typing import Any, Dict, List
+
+from aworld.config.conf import ModelConfig
+from aworld.core.context.processor import CompressionResult, CompressionType, MessagesProcessingResult
+from aworld.core.context.processor.base_compressor import BaseCompressor
+from aworld.logs.util import Color, color_log
+from aworld.models.utils import num_tokens_from_messages
+from aworld.utils import import_package
+
+logger = logging.getLogger(__name__)
+
+
+class TruncateCompressor(BaseCompressor):
+    """
+    Truncate messages compressor for content length management
+    """
+
+    def __init__(self, config: Dict[str, Any] = None, llm_config: ModelConfig = None):
+        super().__init__(config, llm_config)
+        self.model_type = llm_config.model_type if llm_config else "gpt-3.5-turbo"
+        self._init_tokenizer()
+
+    def _init_tokenizer(self):
+        """Initialize tokenizer for text truncation"""
+        try:
+            import_package("tiktoken")
+            import tiktoken
+            
+            if self.model_type.lower() == "qwen":
+                from aworld.models.qwen_tokenizer import qwen_tokenizer
+                self.tokenizer = qwen_tokenizer
+            elif self.model_type.lower() == "openai":
+                from aworld.models.openai_tokenizer import openai_tokenizer
+                self.tokenizer = openai_tokenizer
+            else:
+                try:
+                    self.encoding = tiktoken.encoding_for_model(self.model_type)
+                    self.tokenizer = None  # Use tiktoken directly
+                except KeyError:
+                    logger.warning(f"{self.model_type} model not found. Using cl100k_base encoding.")
+                    self.encoding = tiktoken.get_encoding("cl100k_base")
+                    self.tokenizer = None
+        except ImportError:
+            logger.error("tiktoken not available, text truncation may not work properly")
+            self.tokenizer = None
+            self.encoding = None
+
+    def _count_tokens_from_messages(self, messages: List[Dict[str, Any]]) -> int:
+        """Calculate token count for messages using utils.py method"""
+        return num_tokens_from_messages(messages, model=self.model_type)
+
+    def _count_tokens_from_message(self, msg: Dict[str, Any]) -> int:
+        """Calculate token count for single message using utils.py method"""
+        # Convert single message to list format for num_tokens_from_messages
+        return num_tokens_from_messages([msg], model=self.model_type)
+
+    def _truncate_text(self, text: str, max_tokens: int, keep_both_sides: bool = False) -> str:
+        """Truncate text content using appropriate tokenizer"""
+        if not text:
+            return text
+        
+        # Ensure max_tokens is an integer
+        max_tokens = int(max_tokens)
+        if max_tokens <= 0:
+            return ""
+            
+        try:
+            if self.tokenizer:
+                # Use custom tokenizer (qwen/openai)
+                return self.tokenizer.truncate(text, max_tokens, keep_both_sides=keep_both_sides)
+            elif self.encoding:
+                # Use tiktoken encoding directly
+                tokens = self.encoding.encode(text)
+                if len(tokens) <= max_tokens:
+                    return text
+                
+                if keep_both_sides:
+                    ellipsis = "..."
+                    ellipsis_tokens = self.encoding.encode(ellipsis)
+                    ellipsis_len = len(ellipsis_tokens)
+                    available = max_tokens - ellipsis_len
+                    if available <= 0:
+                        # Not enough space for ellipsis
+                        truncated_tokens = tokens[:max_tokens]
+                    else:
+                        left_len = int(available // 2)
+                        right_len = int(available - left_len)
+                        truncated_tokens = tokens[:left_len] + ellipsis_tokens + tokens[-right_len:]
+                else:
+                    truncated_tokens = tokens[:max_tokens]
+                
+                return self.encoding.decode(truncated_tokens)
+            else:
+                # Fallback: simple character truncation
+                logger.warning("No tokenizer available, using character-based truncation")
+                target_len = max_tokens * 4  # Rough estimate: 1 token = 4 chars
+                target_len = int(target_len)
+                
+                if len(text) <= target_len:
+                    return text
+                
+                if keep_both_sides:
+                    ellipsis = "..."
+                    available = target_len - len(ellipsis)
+                    if available <= 0:
+                        return text[:target_len]
+                    left_len = int(available // 2)
+                    right_len = int(available - left_len)
+                    return text[:left_len] + ellipsis + text[-right_len:]
+                else:
+                    return text[:target_len]
+        except Exception as e:
+            logger.error(f"Text truncation failed: {e}")
+            return text
+
+    def _truncate_message(self, msg: Dict[str, Any], max_tokens: int, keep_both_sides: bool = False):
+        """Truncate single message content"""
+        # Ensure max_tokens is an integer
+        max_tokens = int(max_tokens)
+        
+        content = msg.get("content", "")
+        if isinstance(content, str):
+            truncated_content = self._truncate_text(content, max_tokens, keep_both_sides)
+        else:
+            # Handle complex content formats
+            if isinstance(content, list):
+                text_parts = []
+                for item in content:
+                    if isinstance(item, dict) and item.get("text"):
+                        text_parts.append(item["text"])
+                    elif isinstance(item, str):
+                        text_parts.append(item)
+                if not text_parts:
+                    return None
+                text = '\n'.join(text_parts)
+            else:
+                text = str(content)
+            truncated_content = self._truncate_text(text, max_tokens, keep_both_sides)
+        
+        new_msg = msg.copy()
+        new_msg["content"] = truncated_content
+        return new_msg
+
+    def is_out_of_context(self, messages: List[Dict[str, Any]], max_tokens: int) -> bool:
+        """Check if messages exceed token limit"""
+        max_tokens = int(max_tokens)
+        return self._count_tokens_from_messages(messages) > max_tokens
+
+    def truncate_messages(self, messages: List[Dict[str, Any]], max_tokens: int, 
+                         optimization_enabled: bool = True) -> MessagesProcessingResult:
+        """Truncate messages based on _truncate_input_messages_roughly logic"""
+        start_time = time.time()
+        original_messages_len = len(messages)
+        original_token_len = self._count_tokens_from_messages(messages)
+        
+        # Ensure max_tokens is an integer
+        max_tokens = int(max_tokens)
+        
+        if not optimization_enabled:
+            processing_time = time.time() - start_time
+            return MessagesProcessingResult(
+                original_token_len=original_token_len,
+                processing_token_len=original_token_len,
+                original_messages_len=original_messages_len,
+                processing_messaged_len=original_messages_len,
+                processing_time=processing_time,
+                method_used="no_optimization",
+                processed_messages=messages
+            )
+        
+        if not self.is_out_of_context(messages=messages, max_tokens=max_tokens):
+            processing_time = time.time() - start_time
+            return MessagesProcessingResult(
+                original_token_len=original_token_len,
+                processing_token_len=original_token_len,
+                original_messages_len=original_messages_len,
+                processing_messaged_len=original_messages_len,
+                processing_time=processing_time,
+                method_used="within_context_limit",
+                processed_messages=messages
+            )
+
+        # Group messages by conversation turns
+        turns = []
+        for m in messages:
+            if m.get("role") == "system":
+                continue
+            elif m.get("role") == "user":
+                turns.append([m])
+            else:
+                if turns:
+                    turns[-1].append(m)
+                else:
+                    raise Exception('The input messages (excluding the system message) must start with a user message.')
+
+        # Process system messages
+        if messages and messages[0].get("role") == "system":
+            sys_msg = messages[0]
+            available_token = max_tokens - self._count_tokens_from_message(sys_msg)
+        else:
+            sys_msg = None
+            available_token = max_tokens
+        
+        # Process messages from back to front, keep the latest conversations
+        token_cnt = 0
+        new_messages = []
+        user_message_count = 0
+        for i in range(len(messages) - 1, -1, -1):
+            if messages[i].get("role") == "system":
+                continue
+            
+            cur_token_cnt = self._count_tokens_from_message(messages[i])
+            if cur_token_cnt <= available_token:
+                if messages[i].get("role") == "user":
+                    user_message_count += 1
+                new_messages = [messages[i]] + new_messages
+                available_token -= cur_token_cnt
+            else:
+                # Try to truncate message
+                if (messages[i].get("role") == "user"):
+                    # Truncate user message (not the last one)
+                    color_log(f"to truncate message {messages[i]}", color=Color.pink)
+                    _msg = self._truncate_message(messages[i], max_tokens=int(available_token))
+                    color_log(f"truncated message {messages[i]}, {_msg}", color=Color.pink)
+                    if _msg:
+                        new_messages = [_msg] + new_messages
+                    break
+                elif messages[i].get("role") == "function" or messages[i].get("role") == "assistant" or messages[i].get("role") == "system":
+                    # Truncate function message, keep both ends
+                    logger.debug(f"to truncate message {messages[i]}")
+                    _msg = self._truncate_message(messages[i], max_tokens=int(available_token), keep_both_sides=True)
+                    logger.debug(f"truncated message {messages[i]}, {_msg}")
+                    if _msg:
+                        new_messages = [_msg] + new_messages
+                    # Edge case: if the last message is a very long tool message, it might end up with only system+tool without user message, which will cause LLM call to fail
+                    elif user_message_count == 0:
+                        continue
+                    else:
+                        break
+                else:
+                    # Cannot truncate, record token count and exit
+                    token_cnt = (max_tokens - available_token) + cur_token_cnt
+                    break
+        
+        # Re-add system message
+        if sys_msg is not None:
+            new_messages = [sys_msg] + new_messages
+        
+        # Calculate processed statistics
+        processing_time = time.time() - start_time
+        processing_token_len = self._count_tokens_from_messages(new_messages)
+        processing_messaged_len = len(new_messages)
+        
+        return MessagesProcessingResult(
+            original_token_len=original_token_len,
+            processing_token_len=processing_token_len,
+            original_messages_len=original_messages_len,
+            processing_messaged_len=processing_messaged_len,
+            processing_time=processing_time,
+            method_used="truncate_messages",
+            processed_messages=new_messages
+        )
+
+    def compress(self, content: str) -> CompressionResult:
+        """
+        Compress content by truncating it (for compatibility with BaseCompressor interface)
+        """
+        # This is a simple truncation, not actual compression
+        # For consistency with other compressors, we provide this method
+        original_content = content
+        
+        # Use a reasonable default max_tokens for single content truncation
+        max_tokens = self.config.get("max_tokens", 2000)
+        
+        try:
+            truncated_content = self._truncate_text(content, max_tokens, False)
+            compression_ratio = len(truncated_content) / len(original_content) if original_content else 1.0
+            
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=truncated_content,
+                compression_ratio=compression_ratio,
+                metadata={"method": "truncation", "max_tokens": max_tokens},
+                compression_type=CompressionType.LLM_BASED  # Default type
+            )
+            
+        except Exception as e:
+            logger.error(f"Truncation failed: {e}")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=content,
+                compression_ratio=1.0,
+                metadata={"error": str(e)},
+                compression_type=CompressionType.LLM_BASED
+            )
+
+    def compress_messages(self, messages: List[Dict[str, Any]]) -> CompressionResult:
+        """
+        Compress messages by truncating them (for compatibility with BaseCompressor interface)
+        """
+        if not messages:
+            return CompressionResult(
+                original_content="[]",
+                compressed_content="[]",
+                compression_ratio=1.0,
+                metadata={},
+                compression_type=CompressionType.LLM_BASED
+            )
+        
+        original_content = str(messages)
+        max_tokens = self.config.get("max_tokens", 4000)
+        
+        try:
+            result = self.truncate_messages(messages, max_tokens, optimization_enabled=True)
+            
+            compressed_content = str(result.processed_messages)
+            compression_ratio = result.processing_token_len / result.original_token_len if result.original_token_len > 0 else 1.0
+            
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=compressed_content,
+                compression_ratio=compression_ratio,
+                metadata={
+                    "method": "truncation",
+                    "max_tokens": max_tokens,
+                    "truncated_messages": result.processed_messages,
+                    "original_message_count": result.original_messages_len,
+                    "processed_message_count": result.processing_messaged_len,
+                    "method_used": result.method_used
+                },
+                compression_type=CompressionType.LLM_BASED
+            )
+            
+        except Exception as e:
+            logger.error(f"Message truncation failed: {e}")
+            return CompressionResult(
+                original_content=original_content,
+                compressed_content=original_content,
+                compression_ratio=1.0,
+                metadata={"error": str(e)},
+                compression_type=CompressionType.LLM_BASED
+            )
+
+    def compress_batch(self, contents: List[str]) -> List[CompressionResult]:
+        """Compress multiple contents in batch"""
+        results = []
+        for content in contents:
+            result = self.compress(content)
+            results.append(result)
+        return results