ChatML Guide: Master Structured Prompts for LLMs

# ❌ Fragile approachprompt = "You are a helpful assistant. User: What's the capital of France? Assistant:"

<|im_start|>systemYou are a helpful assistant.<|im_end|><|im_start|>userWhat's the capital of France?<|im_end|><|im_start|>assistant

# Unstructured, fragile promptsprompt = """System: You are helpful.User: HelloAI: Hi there!User: What's the weather?"""

<|im_start|>systemYou are helpful.<|im_end|><|im_start|>userHello<|im_end|><|im_start|>assistantHi there!<|im_end|><|im_start|>userWhat's the weather?<|im_end|><|im_start|>assistant

<|im_start|>  — Marks the beginning of a message<|im_end|>    — Marks the end of a message

<|im_start|>{role}{content}<|im_end|>

<|im_start|>systemYou are a helpful, concise AI assistant specializing in Python programming.<|im_end|><|im_start|>userWrite a function to calculate factorial.<|im_end|><|im_start|>assistantHere's a Python function to calculate factorial:```pythondef factorial(n):    if n == 0 or n == 1:        return 1    return n * factorial(n - 1)```This uses recursion to calculate the factorial efficiently.<|im_end|>

<|im_start|>systemYou are a senior Python developer with 10 years of experience.Guidelines:- Always include error handling- Prioritize code readability- Add docstrings to functions- Suggest performance optimizations when relevantTone: Professional but friendlyFormat: Provide code examples with explanations<|im_end|>

<|im_start|>systemYou are helpful.<|im_end|>

<|im_start|>system# IdentityYou are an expert technical documentation writer for AI/ML projects.# Expertise- 15+ years documenting complex software systems- Specialization in Python, machine learning, and API documentation- Deep understanding of developer workflows# Guidelines1. Use clear, concise language2. Include practical code examples3. Explain "why" not just "how"4. Anticipate edge cases5. Provide troubleshooting tips# Constraints- Never invent functionality that doesn't exist- Always cite sources when referencing external documentation- Avoid jargon without explanation- Don't assume prior knowledge# ToneProfessional, encouraging, and patient. Treat readers as intelligent but learning.# Format- Start with a brief overview- Use code blocks for all examples- Include both simple and advanced examples- End with common pitfalls or tips<|im_end|>

<|im_start|>userSort this list: [3, 1, 4, 1, 5, 9]<|im_end|>

<|im_start|>userI'm building a REST API with Flask for an e-commerce platform. I need to implement rate limiting for my endpoints to prevent abuse.Requirements:- Limit: 100 requests per hour per user- Use Redis for storage (already set up)- Return proper HTTP status codes (429 for rate limit exceeded)- Include Retry-After header in responses- Support both authenticated users (by user_id) and anonymous users (by IP)Current tech stack:- Python 3.11- Flask 3.0- Redis 7.0- Running on AWS EC2Please provide a production-ready implementation with proper error handling.<|im_end|>

<|im_start|>assistantI'll help you implement rate limiting with Flask and Redis. Here's a production-ready solution:## 1. Install Dependencies```bashpip install Flask redis flask-limiter

from flask import Flask, jsonify, requestfrom flask_limiter import Limiterfrom flask_limiter.util import get_remote_addressimport redisapp = Flask(__name__)# Configure Redis connectionredis_client = redis.Redis(    host='localhost',    port=6379,    db=0,    decode_responses=True)# Initialize rate limiterlimiter = Limiter(    app=app,    key_func=get_remote_address,    storage_uri="redis://localhost:6379",    strategy="fixed-window")# Custom key function for authenticated usersdef get_user_identifier():    """Get user ID if authenticated, otherwise use IP address."""    user_id = request.headers.get('X-User-ID')    if user_id:        return f"user:{user_id}"    return f"ip:{get_remote_address()}"# Apply rate limiting to routes@app.route('/api/products')@limiter.limit("100 per hour", key_func=get_user_identifier)def get_products():    return jsonify({"products": [...], "status": "success"})@app.route('/api/orders', methods=['POST'])@limiter.limit("100 per hour", key_func=get_user_identifier)def create_order():    return jsonify({"order_id": "12345", "status": "created"})# Custom error handler@app.errorhandler(429)def ratelimit_handler(e):    return jsonify({        "error": "Rate limit exceeded",        "message": "Too many requests. Please try again later.",        "retry_after": e.description    }), 429if __name__ == '__main__':    app.run(debug=True)

import requests# Test rate limitingfor i in range(105):    response = requests.get('http://localhost:5000/api/products')    print(f"Request {i+1}: {response.status_code}")    if response.status_code == 429:        print(f"Rate limited! Retry after: {response.headers.get('Retry-After')}")        break

<|im_start|>assistantI'll help you implement rate limiting with Flask and Redis. Here's a production-ready solution:## 1. Install Dependencies```bashpip install Flask redis flask-limiter```## 2. Implementation```pythonfrom flask import Flask, jsonify, requestfrom flask_limiter import Limiterfrom flask_limiter.util import get_remote_addressimport redisapp = Flask(__name__)# Configure Redis connectionredis_client = redis.Redis(    host='localhost',    port=6379,    db=0,    decode_responses=True)# Initialize rate limiterlimiter = Limiter(    app=app,    key_func=get_remote_address,    storage_uri="redis://localhost:6379",    strategy="fixed-window")# Custom key function for authenticated usersdef get_user_identifier():    """Get user ID if authenticated, otherwise use IP address."""    user_id = request.headers.get('X-User-ID')    if user_id:        return f"user:{user_id}"    return f"ip:{get_remote_address()}"# Apply rate limiting to routes@app.route('/api/products')@limiter.limit("100 per hour", key_func=get_user_identifier)def get_products():    return jsonify({"products": [...], "status": "success"})@app.route('/api/orders', methods=['POST'])@limiter.limit("100 per hour", key_func=get_user_identifier)def create_order():    return jsonify({"order_id": "12345", "status": "created"})# Custom error handler@app.errorhandler(429)def ratelimit_handler(e):    return jsonify({        "error": "Rate limit exceeded",        "message": "Too many requests. Please try again later.",        "retry_after": e.description    }), 429if __name__ == '__main__':    app.run(debug=True)```## 3. Key Features- **Dual identification**: Uses user ID for authenticated users, IP for anonymous- **Automatic retry-after**: Flask-Limiter adds this header automatically- **Redis persistence**: Survives application restarts- **Custom error handling**: Professional error messages## 4. Testing```pythonimport requests# Test rate limitingfor i in range(105):    response = requests.get('http://localhost:5000/api/products')    print(f"Request {i+1}: {response.status_code}")    if response.status_code == 429:        print(f"Rate limited! Retry after: {response.headers.get('Retry-After')}")        break```## 5. Production Considerations⚠️ **Important notes**:- Adjust limits based on your API capacity- Monitor Redis memory usage- Consider implementing token bucket algorithm for smoother rate limiting- Add logging for rate limit violations- Implement different limits for different endpoints- Consider geographic distribution if using multiple Redis instancesWould you like me to show you how to implement more advanced features like:- Burst allowance- Different limits for premium vs. free users- Rate limit analytics dashboard<|im_end|>

<|im_start|>tool{  "tool_name": "web_search",  "query": "ChatML documentation OpenAI",  "results": [    {      "title": "ChatML Format Specification - OpenAI",      "url": "https://platform.openai.com/docs/guides/chatml",      "snippet": "ChatML is a structured format for representing conversations..."    },    {      "title": "Understanding ChatML - Developer Guide",      "url": "https://example.com/chatml-guide",      "snippet": "Learn how to use ChatML for building production chatbots..."    }  ],  "timestamp": "2025-12-17T10:30:00Z"}<|im_end|>

<|im_start|>userWhat's the current weather in San Francisco?<|im_end|><|im_start|>assistantI'll check the current weather in San Francisco for you.<|im_end|><|im_start|>tool{  "tool_name": "weather_api",  "location": "San Francisco, CA",  "data": {    "temperature": 62,    "condition": "Partly Cloudy",    "humidity": 75,    "wind_speed": 12  }}<|im_end|><|im_start|>assistantThe current weather in San Francisco is:- Temperature: 62°F- Condition: Partly Cloudy- Humidity: 75%- Wind Speed: 12 mphIt's a pleasant day with comfortable temperatures!<|im_end|>

class ChatMLFormatter:    """Production-ready ChatML formatter with validation."""        VALID_ROLES = {'system', 'user', 'assistant', 'tool'}    START_TOKEN = '<|im_start|>'    END_TOKEN = '<|im_end|>'        def __init__(self):        self.messages = []        def add_message(self, role: str, content: str) -> 'ChatMLFormatter':        """Add a message with validation."""        if role not in self.VALID_ROLES:            raise ValueError(f"Invalid role: {role}. Must be one of {self.VALID_ROLES}")                if not content or not content.strip():            raise ValueError("Message content cannot be empty")                self.messages.append({            'role': role,            'content': content.strip()        })        return self  # Enable chaining        def to_chatml(self, include_assistant_start: bool = True) -> str:        """Convert messages to ChatML format."""        chatml = []                for msg in self.messages:            chatml.append(f"{self.START_TOKEN}{msg['role']}")            chatml.append(msg['content'])            chatml.append(self.END_TOKEN)                # Add assistant start token for model completion        if include_assistant_start:            chatml.append(f"{self.START_TOKEN}assistant")                return '\n'.join(chatml)        def from_chatml(self, chatml_string: str) -> 'ChatMLFormatter':        """Parse ChatML string back to messages."""        import re                pattern = rf"{re.escape(self.START_TOKEN)}(\w+)\n(.*?){re.escape(self.END_TOKEN)}"        matches = re.findall(pattern, chatml_string, re.DOTALL)                self.messages = []        for role, content in matches:            if role in self.VALID_ROLES:                self.messages.append({                    'role': role,                    'content': content.strip()                })                return self        def to_dict(self) -> list:        """Convert to OpenAI API format."""        return [{'role': msg['role'], 'content': msg['content']}                 for msg in self.messages]        def __len__(self) -> int:        return len(self.messages)        def __repr__(self) -> str:        return f"ChatMLFormatter({len(self)} messages)"# Usage exampleformatter = ChatMLFormatter()formatter.add_message('system', 'You are a helpful AI assistant.') \         .add_message('user', 'What is ChatML?') \         .add_message('assistant', 'ChatML is a structured format for LLM conversations.')# Generate ChatMLchatml_output = formatter.to_chatml()print(chatml_output)# Convert to OpenAI formatopenai_format = formatter.to_dict()print(openai_format)

import asynciofrom typing import AsyncGeneratorclass StreamingChatML:    """Handle streaming ChatML responses."""        async def stream_response(        self,         messages: list,         model: str = "gpt-4"    ) -> AsyncGenerator[str, None]:        """Stream ChatML formatted responses."""        from openai import AsyncOpenAI                client = AsyncOpenAI()                async for chunk in await client.chat.completions.create(            model=model,            messages=messages,            stream=True        ):            if chunk.choices[0].delta.content:                yield chunk.choices[0].delta.content        async def format_stream(        self,        messages: list    ) -> AsyncGenerator[str, None]:        """Format streaming response as ChatML."""        yield '<|im_start|>assistant\n'                async for token in self.stream_response(messages):            yield token                yield '\n<|im_end|>'# Usageasync def main():    streamer = StreamingChatML()    messages = [        {'role': 'system', 'content': 'You are helpful.'},        {'role': 'user', 'content': 'Count to 5.'}    ]        async for chunk in streamer.format_stream(messages):        print(chunk, end='', flush=True)# Run# asyncio.run(main())

import tiktokenclass ContextWindowManager:    """Manage token limits in ChatML conversations."""        def __init__(self, model: str = "gpt-4", max_tokens: int = 8192):        self.encoding = tiktoken.encoding_for_model(model)        self.max_tokens = max_tokens        def count_tokens(self, messages: list) -> int:        """Count tokens in message list."""        formatter = ChatMLFormatter()        for msg in messages:            formatter.add_message(msg['role'], msg['content'])                chatml_string = formatter.to_chatml()        return len(self.encoding.encode(chatml_string))        def truncate_messages(        self,         messages: list,         reserve_tokens: int = 1000    ) -> list:        """Truncate messages to fit context window."""        available_tokens = self.max_tokens - reserve_tokens                # Always keep system message        result = [messages[0]] if messages[0]['role'] == 'system' else []        current_tokens = self.count_tokens(result)                # Add messages from newest to oldest        for msg in reversed(messages[1:]):            msg_tokens = len(self.encoding.encode(msg['content']))                        if current_tokens + msg_tokens <= available_tokens:                result.insert(1 if result else 0, msg)                current_tokens += msg_tokens            else:                break                return result# Usagemanager = ContextWindowManager(model="gpt-4", max_tokens=8192)long_conversation = [    {'role': 'system', 'content': 'You are helpful.'},    # ... many messages ...]optimized = manager.truncate_messages(long_conversation, reserve_tokens=500)print(f"Reduced from {len(long_conversation)} to {len(optimized)} messages")

def format_for_openai(messages: list) -> str:    """Direct ChatML format for OpenAI."""    formatter = ChatMLFormatter()    for msg in messages:        formatter.add_message(msg['role'], msg['content'])    return formatter.to_chatml()

def format_for_claude(messages: list) -> str:    """Convert ChatML to Claude's format."""    claude_prompt = ""        for msg in messages:        if msg['role'] == 'system':            claude_prompt += f"\n\nSystem: {msg['content']}"        elif msg['role'] == 'user':            claude_prompt += f"\n\nHuman: {msg['content']}"        elif msg['role'] == 'assistant':            claude_prompt += f"\n\nAssistant: {msg['content']}"        claude_prompt += "\n\nAssistant:"    return claude_prompt

def format_for_qwen(messages: list) -> str:    """Qwen uses identical ChatML format."""    return format_for_openai(messages)  # Same format!

class UniversalChatMLAdapter:    """Adapt ChatML for any LLM."""        ADAPTERS = {        'openai': lambda msgs: ChatMLFormatter().from_dict(msgs).to_chatml(),        'claude': format_for_claude,        'qwen': format_for_qwen,        # Add more as needed    }        def format(self, messages: list, target: str) -> str:        """Format messages for target LLM."""        if target not in self.ADAPTERS:            raise ValueError(f"No adapter for {target}")                return self.ADAPTERS[target](messages)# Usageadapter = UniversalChatMLAdapter()messages = [    {'role': 'system', 'content': 'You are helpful.'},    {'role': 'user', 'content': 'Hello!'}]# Format for different modelsopenai_format = adapter.format(messages, 'openai')claude_format = adapter.format(messages, 'claude')qwen_format = adapter.format(messages, 'qwen')

class ConversationTemplate:    """Reusable conversation templates."""        TEMPLATES = {        'code_review': [            {                'role': 'system',                'content': '''You are an expert code reviewer.Guidelines:- Focus on security vulnerabilities- Check for performance issues- Verify error handling- Assess code readability'''            }        ],        'technical_writer': [            {                'role': 'system',                'content': '''You are a technical documentation expert.Style:- Use clear, concise language- Include code examples- Add practical use cases- Provide warnings for edge cases'''            }        ]    }        @classmethod    def create(cls, template_name: str, user_message: str) -> list:        """Create conversation from template."""        if template_name not in cls.TEMPLATES:            raise ValueError(f"Unknown template: {template_name}")                messages = cls.TEMPLATES[template_name].copy()        messages.append({'role': 'user', 'content': user_message})        return messages# Usagemessages = ConversationTemplate.create(    'code_review',    'Review this function: def add(a, b): return a + b')

from datetime import datetimefrom typing import Optionalimport jsonclass ConversationState:    """Maintain conversation state with metadata."""        def __init__(self, conversation_id: str):        self.conversation_id = conversation_id        self.messages = []        self.metadata = {            'created_at': datetime.utcnow().isoformat(),            'updated_at': datetime.utcnow().isoformat(),            'turn_count': 0        }        def add_turn(        self,         user_message: str,         assistant_response: str,        metadata: Optional[dict] = None    ):        """Add a complete conversation turn."""        self.messages.extend([            {                'role': 'user',                'content': user_message,                'timestamp': datetime.utcnow().isoformat()            },            {                'role': 'assistant',                'content': assistant_response,                'timestamp': datetime.utcnow().isoformat()            }        ])                self.metadata['turn_count'] += 1        self.metadata['updated_at'] = datetime.utcnow().isoformat()                if metadata:            self.metadata.update(metadata)        def to_chatml(self) -> str:        """Convert to ChatML format."""        formatter = ChatMLFormatter()        for msg in self.messages:            formatter.add_message(msg['role'], msg['content'])        return formatter.to_chatml()        def save(self, filepath: str):        """Persist conversation state."""        state = {            'conversation_id': self.conversation_id,            'messages': self.messages,            'metadata': self.metadata        }                with open(filepath, 'w') as f:            json.dump(state, f, indent=2)        @classmethod    def load(cls, filepath: str) -> 'ConversationState':        """Load conversation state."""        with open(filepath, 'r') as f:            state = json.load(f)                conversation = cls(state['conversation_id'])        conversation.messages = state['messages']        conversation.metadata = state['metadata']        return conversation# Usageconversation = ConversationState('conv_001')conversation.add_turn(    user_message="What is ChatML?",    assistant_response="ChatML is a structured format...",    metadata={'model': 'gpt-4', 'tokens': 150})conversation.save('conversation_001.json')

class SecureChatMLFormatter(ChatMLFormatter):    """ChatML formatter with role-based access control."""        ALLOWED_ROLES = {        'admin': {'system', 'user', 'assistant', 'tool'},        'developer': {'user', 'assistant', 'tool'},        'user': {'user'}    }        def __init__(self, user_role: str = 'user'):        super().__init__()        self.user_role = user_role        def add_message(self, role: str, content: str) -> 'SecureChatMLFormatter':        """Add message with permission check."""        if role not in self.ALLOWED_ROLES.get(self.user_role, set()):            raise PermissionError(                f"Role '{self.user_role}' cannot add '{role}' messages"            )                return super().add_message(role, content)# Usageadmin_formatter = SecureChatMLFormatter(user_role='admin')admin_formatter.add_message('system', 'You are helpful.')  # ✅ Alloweduser_formatter = SecureChatMLFormatter(user_role='user')# user_formatter.add_message('system', 'Hack!')  # ❌ PermissionError

import refrom typing import List, Dict, Tupleclass ChatMLValidator:    """Validate ChatML inputs for production."""        # Dangerous patterns to block    DANGEROUS_PATTERNS = [        r'<\|im_start\|>',  # Injection attempts        r'<\|im_end\|>',        r'<script>',        # XSS attempts        r'javascript:',        r'data:text/html'    ]        MAX_MESSAGE_LENGTH = 10000    MAX_MESSAGES = 100        @classmethod    def validate_message(cls, role: str, content: str) -> Tuple[bool, str]:        """Validate a single message."""        # Check role        if role not in ChatMLFormatter.VALID_ROLES:            return False, f"Invalid role: {role}"                # Check length        if len(content) > cls.MAX_MESSAGE_LENGTH:            return False, f"Message too long: {len(content)} > {cls.MAX_MESSAGE_LENGTH}"                # Check for injection attempts        for pattern in cls.DANGEROUS_PATTERNS:            if re.search(pattern, content, re.IGNORECASE):                return False, f"Dangerous pattern detected: {pattern}"                return True, "Valid"        @classmethod    def validate_conversation(cls, messages: List[Dict]) -> Tuple[bool, str]:        """Validate entire conversation."""        if len(messages) > cls.MAX_MESSAGES:            return False, f"Too many messages: {len(messages)} > {cls.MAX_MESSAGES}"                for i, msg in enumerate(messages):            valid, error = cls.validate_message(msg['role'], msg['content'])            if not valid:                return False, f"Message {i}: {error}"                return True, "Valid"# Usagevalidator = ChatMLValidator()messages = [    {'role': 'user', 'content': 'Hello!'},    {'role': 'assistant', 'content': 'Hi there!'}]valid, message = validator.validate_conversation(messages)if not valid:    print(f"Validation failed: {message}")

from tenacity import retry, stop_after_attempt, wait_exponentialclass RobustChatMLClient:    """Production ChatML client with error handling."""        def __init__(self, api_key: str):        from openai import OpenAI        self.client = OpenAI(api_key=api_key)        @retry(        stop=stop_after_attempt(3),        wait=wait_exponential(multiplier=1, min=2, max=10)    )    def generate_response(        self,         messages: list,        model: str = "gpt-4",        **kwargs    ) -> dict:        """Generate response with automatic retries."""        try:            # Validate input            valid, error = ChatMLValidator.validate_conversation(messages)            if not valid:                raise ValueError(f"Invalid conversation: {error}")                        # Make API call            response = self.client.chat.completions.create(                model=model,                messages=messages,                **kwargs            )                        return {                'success': True,                'content': response.choices[0].message.content,                'model': response.model,                'tokens': response.usage.total_tokens            }                    except Exception as e:            return {                'success': False,                'error': str(e),                'error_type': type(e).__name__            }# Usageclient = RobustChatMLClient(api_key="your-key")result = client.generate_response([    {'role': 'user', 'content': 'Hello!'}])if result['success']:    print(result['content'])else:    print(f"Error: {result['error']}")

import timefrom collections import dequefrom threading import Lockclass RateLimiter:    """Token bucket rate limiter for ChatML requests."""        def __init__(self, requests_per_minute: int = 60):        self.rpm = requests_per_minute        self.requests = deque()        self.lock = Lock()        def acquire(self) -> bool:        """Acquire permission to make a request."""        with self.lock:            now = time.time()                        # Remove requests older than 1 minute            while self.requests and self.requests[0] < now - 60:                self.requests.popleft()                        # Check if we can make request            if len(self.requests) < self.rpm:                self.requests.append(now)                return True                        return False        def wait_if_needed(self):        """Block until request can be made."""        while not self.acquire():            time.sleep(0.1)# Usagelimiter = RateLimiter(requests_per_minute=60)for i in range(100):    limiter.wait_if_needed()    # Make API call    print(f"Request {i+1}")

import loggingfrom datetime import datetimeimport jsonclass ChatMLLogger:    """Comprehensive logging for ChatML operations."""        def __init__(self, log_file: str = 'chatml.log'):        self.logger = logging.getLogger('ChatML')        self.logger.setLevel(logging.INFO)                handler = logging.FileHandler(log_file)        handler.setFormatter(logging.Formatter(            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'        ))        self.logger.addHandler(handler)        def log_request(self, messages: list, metadata: dict = None):        """Log ChatML request."""        self.logger.info(json.dumps({            'event': 'request',            'timestamp': datetime.utcnow().isoformat(),            'message_count': len(messages),            'metadata': metadata or {}        }))        def log_response(self, response: dict, metadata: dict = None):        """Log ChatML response."""        self.logger.info(json.dumps({            'event': 'response',            'timestamp': datetime.utcnow().isoformat(),            'success': response.get('success', False),            'tokens': response.get('tokens', 0),            'metadata': metadata or {}        }))        def log_error(self, error: Exception, context: dict = None):        """Log errors with context."""        self.logger.error(json.dumps({            'event': 'error',            'timestamp': datetime.utcnow().isoformat(),            'error_type': type(error).__name__,            'error_message': str(error),            'context': context or {}        }))# Usagelogger = ChatMLLogger()messages = [{'role': 'user', 'content': 'Hello'}]logger.log_request(messages, {'user_id': 'user_123'})

def verify_tokenization(text: str, model: str = "gpt-4") -> None:    """Verify ChatML tokens are properly recognized."""    import tiktoken        encoding = tiktoken.encoding_for_model(model)    tokens = encoding.encode(text)        # Check if special tokens are single tokens    im_start_tokens = encoding.encode('<|im_start|>')    im_end_tokens = encoding.encode('<|im_end|>')        print(f"<|im_start|> tokens: {len(im_start_tokens)}")    print(f"<|im_end|> tokens: {len(im_end_tokens)}")        if len(im_start_tokens) != 1 or len(im_end_tokens) != 1:        print("⚠️ Warning: Special tokens not recognized as single tokens")        print("Solution: Ensure you're using a ChatML-compatible model")verify_tokenization('<|im_start|>system\nHello<|im_end|>')

class ContextPreserver:    """Preserve important context across long conversations."""        def __init__(self, max_context_messages: int = 10):        self.max_context = max_context_messages        self.important_indices = set()        def mark_important(self, index: int):        """Mark a message as important (always keep)."""        self.important_indices.add(index)        def compress_messages(self, messages: list) -> list:        """Compress messages while preserving important ones."""        if len(messages) <= self.max_context:            return messages                # Always keep system message        result = [messages[0]] if messages[0]['role'] == 'system' else []                # Keep important messages        for idx in sorted(self.important_indices):            if idx < len(messages):                result.append(messages[idx])                # Fill remaining slots with recent messages        recent_count = self.max_context - len(result)        result.extend(messages[-recent_count:])                return result# Usagepreserver = ContextPreserver(max_context_messages=10)preserver.mark_important(2)  # Keep message at index 2compressed = preserver.compress_messages(long_conversation)

def validate_chatml_syntax(chatml_string: str) -> Tuple[bool, List[str]]:    """Validate ChatML syntax."""    errors = []        # Check matching start/end tokens    start_count = chatml_string.count('<|im_start|>')    end_count = chatml_string.count('<|im_end|>')        if start_count != end_count:        errors.append(f"Mismatched tokens: {start_count} starts, {end_count} ends")        # Check role validity    import re    roles = re.findall(r'<\|im_start\|>(\w+)', chatml_string)    valid_roles = {'system', 'user', 'assistant', 'tool'}        for role in roles:        if role not in valid_roles:            errors.append(f"Invalid role: {role}")        # Check empty messages    messages = re.findall(        r'<\|im_start\|>\w+\n(.*?)<\|im_end\|>',        chatml_string,        re.DOTALL    )        for i, msg in enumerate(messages):        if not msg.strip():            errors.append(f"Empty message at position {i}")        return len(errors) == 0, errors# Usagechatml = "<|im_start|>system\nHello<|im_end|>"valid, errors = validate_chatml_syntax(chatml)if not valid:    print("Validation errors:")    for error in errors:        print(f"  - {error}")

import functoolsimport time# 1. Caching@functools.lru_cache(maxsize=128)def cached_format(messages_tuple: tuple) -> str:    """Cache formatted ChatML strings."""    messages = list(messages_tuple)    formatter = ChatMLFormatter()    for msg in messages:        formatter.add_message(msg['role'], msg['content'])    return formatter.to_chatml()# 2. Connection Poolingfrom openai import OpenAIclass ConnectionPool:    """Manage OpenAI client connections."""        def __init__(self, api_key: str, pool_size: int = 5):        self.clients = [OpenAI(api_key=api_key) for _ in range(pool_size)]        self.current = 0        def get_client(self) -> OpenAI:        """Get next available client (round-robin)."""        client = self.clients[self.current]        self.current = (self.current + 1) % len(self.clients)        return client# 3. Batch Processingclass BatchProcessor:    """Process multiple ChatML requests efficiently."""        def __init__(self, batch_size: int = 10):        self.batch_size = batch_size        self.queue = []        async def add_request(self, messages: list):        """Add request to batch queue."""        self.queue.append(messages)                if len(self.queue) >= self.batch_size:            await self.process_batch()        async def process_batch(self):        """Process accumulated requests."""        # Process all queued requests        results = []        for messages in self.queue:            # Make API call            result = await self.call_api(messages)            results.append(result)                self.queue.clear()        return results

<!-- Critic role for self-evaluation --><|im_start|>criticLet me evaluate the previous response:- Code quality: 8/10- Completeness: 9/10- Error handling: 7/10Suggestions: Add input validation<|im_end|><!-- Planner role for multi-step reasoning --><|im_start|>plannerTask breakdown:1. Parse user requirements2. Research available APIs3. Design architecture4. Implement solution5. Test and validate<|im_end|><!-- Observer role for monitoring --><|im_start|>observerMonitoring conversation health:- Token usage: 1,234 / 8,192 (15%)- Turn count: 5- Average response time: 2.3s- User satisfaction: High (inferred)<|im_end|>

<|im_start|>usercontent: "What's the weather in New York?"metadata:  timestamp: "2025-12-17T10:30:00Z"  location: "New York, NY"  user_id: "user_123"  device: "mobile"  session_id: "sess_abc"  intent: "weather_query"  priority: "normal"<|im_end|>

<|im_start|>assistantI'll break this complex task into subtasks:<|im_start|>plannerPrimary task: Build REST APISubtasks:1. Design database schema2. Implement authentication3. Create CRUD endpoints<|im_end|>Let me start with subtask 1...<|im_start|>assistantFor the database schema, I recommend:[detailed response]<|im_end|><|im_end|>

<|im_start|>assistant{  "response_type": "structured",  "sections": [    {      "heading": "Solution Overview",      "content": "Here's how to implement rate limiting..."    },    {      "heading": "Code Implementation",      "content": "[code block]",      "language": "python"    },    {      "heading": "Testing Strategy",      "content": "Run these tests..."    }  ],  "confidence": 0.95,  "sources_cited": 3}<|im_end|>

|im_start|>userこんにちは！ChatMLについて教えてください。<|im_end|><|im_start|>assistantChatMLは、大規模言語モデルの会話を構造化するためのフォーマットです...<|im_end|>

# 1. Syntax validationvalid, errors = validate_chatml_syntax(chatml_string)# 2. Token verificationverify_tokenization(chatml_string, model="gpt-4")# 3. Message validationvalid, error = ChatMLValidator.validate_message(role, content)

conversation_metadata = {    'format_version': '1.0',    'chatml_spec': '2024-01',    'created_at': '2025-12-17T10:30:00Z',    'model': 'gpt-4',    'app_version': '2.1.0'}

<|im_start|>usercontent: "Analyze this image"attachments:  - type: "image"    url: "https://example.com/image.jpg"    description: "Product photo"    mime_type: "image/jpeg"  - type: "document"    url: "https://example.com/doc.pdf"    description: "Technical specifications"    mime_type: "application/pdf"<|im_end|>