OpenAI Assistants API: The Complete Guide to Intelligent Agents

User Input → Thread Update → Run Creation → Assistant Processing → Tool Execution → Response Generation → Thread Update

# Weather service integration example
{
    "type": "function",
    "function": {
        "name": "get_weather_forecast",
        "description": "Retrieve detailed weather forecast for any location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name or geographic coordinates"
                },
                "forecast_days": {
                    "type": "integer",
                    "description": "Number of days to forecast (1-7)",
                    "minimum": 1,
                    "maximum": 7
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature units"
                }
            },
            "required": ["location"]
        }
    }
}

# Streaming implementation for real-time UI updates
run = client.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
    stream=True
)

for event in run:
    if event.event == 'thread.message.delta':
        # Update UI with streaming content
        update_chat_interface(event.data.delta.content)

# Context-aware tool selection
def get_tools_for_context(user_role, conversation_topic):
    base_tools = [{"type": "file_search"}]
    
    if user_role == "data_analyst":
        base_tools.append({"type": "code_interpreter"})
    
    if conversation_topic == "weather":
        base_tools.append({
            "type": "function",
            "function": weather_function_schema
        })
    
    return base_tools

# Apply context-specific tools
run = client.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
    tools=get_tools_for_context(user.role, conversation.topic)
)

# Essential imports for robust implementation
from openai import OpenAI
import time
import json
import asyncio
from typing import Dict, List, Optional, Any
import logging
from datetime import datetime

# Configure logging for debugging and monitoring
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize client with proper error handling
try:
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        timeout=30.0,  # Prevent hanging requests
        max_retries=3   # Automatic retry on failures
    )
except Exception as e:
    logger.error(f"Failed to initialize OpenAI client: {e}")
    raise

def create_specialized_assistant(domain: str, capabilities: List[str]) -> Any:
    """
    Create domain-specific assistant with tailored instructions and tools
    """
    
    # Domain-specific instruction templates
    instructions = {
        "data_analysis": """
        You are an expert data analyst with deep expertise in statistical analysis, 
        data visualization, and machine learning. Your approach is methodical and 
        evidence-based. Always:
        
        1. Explain your analytical approach before diving into code
        2. Provide clear interpretations of results
        3. Suggest actionable insights based on findings
        4. Validate assumptions and highlight limitations
        
        When working with data, prioritize accuracy and reproducibility.
        """,
        
        "customer_support": """
        You are a knowledgeable customer support specialist with access to 
        comprehensive product documentation. Your communication style is:
        
        - Professional yet approachable
        - Clear and concise
        - Solution-focused
        - Empathetic to customer concerns
        
        Always search the knowledge base before providing answers, and escalate 
        complex issues to human agents when appropriate.
        """,
        
        "technical_writer": """
        You are a skilled technical writer who creates clear, comprehensive 
        documentation. Your writing follows these principles:
        
        - Start with the user's perspective and goals
        - Use active voice and clear, concise language
        - Provide practical examples and code samples
        - Structure information logically with proper headings
        - Include troubleshooting tips and common pitfalls
        """
    }
    
    # Map capabilities to tools
    tool_mapping = {
        "code_execution": {"type": "code_interpreter"},
        "document_search": {"type": "file_search"},
        "api_integration": {
            "type": "function",
            "function": {
                "name": "api_request",
                "description": "Make HTTP requests to external APIs",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "url": {"type": "string"},
                        "method": {"type": "string", "enum": ["GET", "POST", "PUT", "DELETE"]},
                        "headers": {"type": "object"},
                        "data": {"type": "object"}
                    },
                    "required": ["url", "method"]
                }
            }
        }
    }
    
    # Select appropriate tools
    tools = [tool_mapping[cap] for cap in capabilities if cap in tool_mapping]
    
    try:
        assistant = client.assistants.create(
            name=f"{domain.title().replace('_', ' ')} Assistant",
            instructions=instructions.get(domain, "You are a helpful AI assistant."),
            model="gpt-4o",  # Use latest model for best performance
            tools=tools,
            temperature=0.1,  # Lower temperature for consistent, focused responses
            metadata={
                "domain": domain,
                "created_date": datetime.now().isoformat(),
                "capabilities": ",".join(capabilities)
            }
        )
        
        logger.info(f"Created assistant: {assistant.id} for domain: {domain}")
        return assistant
        
    except Exception as e:
        logger.error(f"Failed to create assistant: {e}")
        raise

class KnowledgeBaseManager:
    """
    Manages file uploads, vector stores, and knowledge base operations
    """
    
    def __init__(self, client: OpenAI):
        self.client = client
        self.upload_cache = {}
    
    def upload_document(self, file_path: str, metadata: Dict[str, Any] = None) -> str:
        """
        Upload document with caching and error handling
        """
        
        # Check cache to avoid duplicate uploads
        file_hash = self._calculate_file_hash(file_path)
        if file_hash in self.upload_cache:
            logger.info(f"Using cached file: {file_path}")
            return self.upload_cache[file_hash]
        
        try:
            with open(file_path, "rb") as file:
                uploaded_file = self.client.files.create(
                    file=file,
                    purpose="assistants"
                )
            
            # Cache the result
            self.upload_cache[file_hash] = uploaded_file.id
            
            logger.info(f"Uploaded file: {file_path} -> {uploaded_file.id}")
            return uploaded_file.id
            
        except Exception as e:
            logger.error(f"Failed to upload {file_path}: {e}")
            raise
    
    def create_vector_store(self, name: str, file_paths: List[str]) -> str:
        """
        Create optimized vector store for document retrieval
        """
        
        # Upload all files
        file_ids = []
        for path in file_paths:
            try:
                file_id = self.upload_document(path)
                file_ids.append(file_id)
            except Exception as e:
                logger.warning(f"Skipping file {path}: {e}")
                continue
        
        if not file_ids:
            raise ValueError("No files successfully uploaded")
        
        # Create vector store
        vector_store = self.client.vector_stores.create(
            name=name,
            file_ids=file_ids,
            metadata={
                "created_date": datetime.now().isoformat(),
                "file_count": len(file_ids)
            }
        )
        
        logger.info(f"Created vector store: {vector_store.id} with {len(file_ids)} files")
        return vector_store.id
    
    def _calculate_file_hash(self, file_path: str) -> str:
        """Calculate hash for file caching"""
        import hashlib
        with open(file_path, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()

class ConversationManager:
    """
    Manages conversation threads and run execution with advanced error handling
    """
    
    def __init__(self, client: OpenAI):
        self.client = client
        self.active_runs = {}
        self.thread_cache = {}
    
    async def process_conversation(
        self, 
        assistant_id: str, 
        user_message: str, 
        thread_id: Optional[str] = None,
        user_files: List[str] = None
    ) -> Dict[str, Any]:
        """
        Process a conversation with comprehensive error handling and monitoring
        """
        
        try:
            # Get or create thread
            if thread_id and thread_id in self.thread_cache:
                thread = self.thread_cache[thread_id]
            else:
                thread = self.client.threads.create()
                self.thread_cache[thread.id] = thread
            
            # Add user message with optional file attachments
            message_params = {
                "thread_id": thread.id,
                "role": "user",
                "content": user_message
            }
            
            if user_files:
                message_params["attachments"] = [
                    {"file_id": file_id, "tools": [{"type": "file_search"}]}
                    for file_id in user_files
                ]
            
            self.client.threads.messages.create(**message_params)
            
            # Execute run with monitoring
            run = self.client.threads.runs.create(
                thread_id=thread.id,
                assistant_id=assistant_id
            )
            
            # Track active run
            self.active_runs[run.id] = {
                "start_time": time.time(),
                "thread_id": thread.id,
                "status": "started"
            }
            
            # Execute with comprehensive monitoring
            result = await self._execute_run_with_monitoring(run, thread.id)
            
            return {
                "thread_id": thread.id,
                "run_id": run.id,
                "response": result,
                "status": "completed"
            }
            
        except Exception as e:
            logger.error(f"Conversation processing failed: {e}")
            return {
                "thread_id": thread_id,
                "error": str(e),
                "status": "failed"
            }
    
    async def _execute_run_with_monitoring(self, run: Any, thread_id: str) -> str:
        """
        Execute run with comprehensive monitoring and tool handling
        """
        
        start_time = time.time()
        timeout = 300  # 5 minutes timeout
        
        while run.status in ["queued", "in_progress", "requires_action"]:
            # Check timeout
            if time.time() - start_time > timeout:
                logger.error(f"Run {run.id} timed out after {timeout} seconds")
                raise TimeoutError("Run execution timeout")
            
            # Handle tool calls
            if run.status == "requires_action":
                await self._handle_tool_calls(run, thread_id)
            
            # Wait before polling again
            await asyncio.sleep(1)
            
            # Retrieve updated run status
            run = self.client.threads.runs.retrieve(
                thread_id=thread_id,
                run_id=run.id
            )
            
            # Update monitoring
            self.active_runs[run.id]["status"] = run.status
        
        # Clean up monitoring
        if run.id in self.active_runs:
            del self.active_runs[run.id]
        
        # Handle completion
        if run.status == "completed":
            messages = self.client.threads.messages.list(
                thread_id=thread_id,
                order="desc",
                limit=1
            )
            
            if messages.data:
                return messages.data[0].content[0].text.value
            
        elif run.status == "failed":
            logger.error(f"Run {run.id} failed: {run.last_error}")
            raise RuntimeError(f"Run failed: {run.last_error}")
        
        return "No response generated"
    
    async def _handle_tool_calls(self, run: Any, thread_id: str):
        """
        Handle tool calls with proper error handling and logging
        """
        
        if not run.required_action or not run.required_action.submit_tool_outputs:
            return
        
        tool_outputs = []
        
        for tool_call in run.required_action.submit_tool_outputs.tool_calls:
            logger.info(f"Executing tool: {tool_call.function.name}")
            
            try:
                # Parse function arguments
                args = json.loads(tool_call.function.arguments)
                
                # Execute function (this would be your custom function registry)
                result = await self._execute_function(
                    tool_call.function.name, 
                    args
                )
                
                tool_outputs.append({
                    "tool_call_id": tool_call.id,
                    "output": json.dumps(result) if isinstance(result, dict) else str(result)
                })
                
            except Exception as e:
                logger.error(f"Tool execution failed: {e}")
                tool_outputs.append({
                    "tool_call_id": tool_call.id,
                    "output": f"Error: {str(e)}"
                })
        
        # Submit tool outputs
        if tool_outputs:
            self.client.threads.runs.submit_tool_outputs(
                thread_id=thread_id,
                run_id=run.id,
                tool_outputs=tool_outputs
            )
    
    async def _execute_function(self, function_name: str, args: Dict[str, Any]) -> Any:
        """
        Execute custom functions - implement your function registry here
        """
        
        # Example function registry
        functions = {
            "get_weather": self._get_weather,
            "api_request": self._make_api_request,
            "database_query": self._query_database
        }
        
        if function_name in functions:
            return await functions[function_name](**args)
        else:
            raise ValueError(f"Unknown function: {function_name}")
    
    async def _get_weather(self, location: str, units: str = "celsius") -> Dict[str, Any]:
        """Example weather function implementation"""
        # Implement your weather API integration here
        return {
            "location": location,
            "temperature": 22,
            "units": units,
            "conditions": "sunny"
        }
    
    async def _make_api_request(self, url: str, method: str, headers: Dict = None, data: Dict = None) -> Dict[str, Any]:
        """Example API request function"""
        # Implement your API request logic here
        return {"status": "success", "data": "API response"}
    
    async def _query_database(self, query: str, parameters: List = None) -> List[Dict[str, Any]]:
        """Example database query function"""
        # Implement your database integration here
        return [{"id": 1, "result": "Database result"}]

async def main():
    """
    Complete example demonstrating the Assistants API workflow
    """
    
    # Initialize managers
    kb_manager = KnowledgeBaseManager(client)
    conv_manager = ConversationManager(client)
    
    # Create knowledge base
    vector_store_id = kb_manager.create_vector_store(
        name="Product Documentation",
        file_paths=["docs/user_guide.pdf", "docs/api_reference.txt", "docs/faq.md"]
    )
    
    # Create specialized assistant
    assistant = create_specialized_assistant(
        domain="customer_support",
        capabilities=["document_search", "api_integration"]
    )
    
    # Associate knowledge base with assistant
    client.assistants.update(
        assistant_id=assistant.id,
        tool_resources={
            "file_search": {"vector_store_ids": [vector_store_id]}
        }
    )
    
    # Process customer inquiry
    result = await conv_manager.process_conversation(
        assistant_id=assistant.id,
        user_message="How do I reset my password? I can't find the option in my account settings.",
        thread_id=None  # Creates new thread
    )
    
    print(f"Assistant Response: {result['response']}")
    
    # Continue conversation
    follow_up = await conv_manager.process_conversation(
        assistant_id=assistant.id,
        user_message="What if I don't receive the reset email?",
        thread_id=result['thread_id']  # Continue same thread
    )
    
    print(f"Follow-up Response: {follow_up['response']}")

# Run the example
if __name__ == "__main__":
    asyncio.run(main())

class RobustAssistantManager:
    """
    Production-ready assistant manager with comprehensive error handling
    """
    
    def __init__(self, client: OpenAI):
        self.client = client
        self.retry_config = {
            "max_retries": 3,
            "base_delay": 1.0,
            "max_delay": 60.0,
            "exponential_backoff": True
        }
    
    async def execute_with_resilience(
        self, 
        operation: callable, 
        *args, 
        **kwargs
    ) -> Any:
        """
        Execute operations with exponential backoff and intelligent retry logic
        """
        
        last_exception = None
        
        for attempt in range(self.retry_config["max_retries"]):
            try:
                return await operation(*args, **kwargs)
                
            except openai.RateLimitError as e:
                # Handle rate limiting with smart backoff
                delay = self._calculate_backoff_delay(attempt, e.retry_after)
                logger.warning(f"Rate limited, retrying in {delay}s: {e}")
                await asyncio.sleep(delay)
                last_exception = e
                
            except openai.APIConnectionError as e:
                # Handle network issues
                delay = self._calculate_backoff_delay(attempt)
                logger.warning(f"Connection error, retrying in {delay}s: {e}")
                await asyncio.sleep(delay)
                last_exception = e
                
            except openai.APIError as e:
                # Handle API errors based on status code
                if e.status_code >= 500:
                    # Server errors - retry
                    delay = self._calculate_backoff_delay(attempt)
                    logger.warning(f"Server error, retrying in {delay}s: {e}")
                    await asyncio.sleep(delay)
                    last_exception = e
                else:
                    # Client errors - don't retry
                    logger.error(f"Client error (no retry): {e}")
                    raise
                    
            except Exception as e:
                # Unexpected errors
                logger.error(f"Unexpected error: {e}")
                raise
        
        # All retries exhausted
        logger.error(f"Operation failed after {self.retry_config['max_retries']} attempts")
        raise last_exception
    
    def _calculate_backoff_delay(self, attempt: int, suggested_delay: float = None) -> float:
        """
        Calculate intelligent backoff delay
        """
        
        if suggested_delay:
            return min(suggested_delay, self.retry_config["max_delay"])
        
        if self.retry_config["exponential_backoff"]:
            delay = self.retry_config["base_delay"] * (2 ** attempt)
        else:
            delay = self.retry_config["base_delay"]
        
        # Add jitter to prevent thundering herd
        import random
        jitter = random.uniform(0.1, 0.3) * delay
        
        return min(delay + jitter, self.retry_config["max_delay"])

import asyncio
import hashlib
import logging
from typing import Any, Dict, Optional

import openai
from openai import OpenAI

logger = logging.getLogger(__name__)


class FallbackAssistantService:
    """
    Implements fallback strategies for AI service failures.  
    失效优雅降级：⭢ 主 Assistant → 备用 Assistant → 降级策略
    """

    # --------------------------- 初始化 --------------------------- #

    def __init__(self, primary_client: OpenAI, fallback_client: Optional[OpenAI] = None):
        self.primary = primary_client
        self.fallback = fallback_client

        # 简易内存缓存（生产建议接入 Redis 或 DynamoDB 等外部缓存）
        self._local_cache: Dict[str, str] = {}

        # 可插拔的降级策略
        self.fallback_strategies = {
            "simple_response": self._simple_fallback,
            "template_response": self._template_fallback,
            "cached_response": self._cached_fallback,
        }

    # ------------------------ 公共入口 ------------------------ #

    async def get_response_with_fallback(
        self,
        assistant_id: str,
        user_message: str,
        fallback_strategy: str = "template_response",
    ) -> Dict[str, Any]:
        """
        Try primary → fallback assistant → graceful degradation strategy.
        """

        cache_key = self._cache_key(user_message)

        # ---------- Primary Assistant ---------- #
        try:
            response = await self._execute_primary_assistant(assistant_id, user_message)
            # 缓存成功结果，便于后续命中
            self._cache_set(cache_key, response)
            return {"response": response, "source": "primary", "fallback_used": False}

        except Exception as primary_error:
            logger.warning(f"Primary assistant failed: {primary_error!r}")

        # ---------- Fallback Assistant ---------- #
        if self.fallback:
            try:
                response = await self._execute_fallback_assistant(user_message)
                # 缓存成功结果
                self._cache_set(cache_key, response)
                return {
                    "response": response,
                    "source": "fallback_assistant",
                    "fallback_used": True,
                }
            except Exception as fallback_error:
                logger.warning(f"Fallback assistant failed: {fallback_error!r}")

        # ---------- Graceful Degradation ---------- #
        if fallback_strategy in self.fallback_strategies:
            response = await self.fallback_strategies[fallback_strategy](user_message)
            return {
                "response": response,
                "source": f"fallback_{fallback_strategy}",
                "fallback_used": True,
            }

        # ---------- Last Resort ---------- #
        return {
            "response": "I'm experiencing technical difficulties. Please try again later.",
            "source": "error_fallback",
            "fallback_used": True,
        }

    # -------------------- 降级策略实现 -------------------- #

    async def _simple_fallback(self, user_message: str) -> str:
        """
        最简降级：直接返回通用占位文本。
        """
        return (
            "⚠️ Our intelligent response system is temporarily unavailable. "
            "However, your request has been queued and will be processed as soon as possible."
        )

    async def _template_fallback(self, user_message: str) -> str:
        """
        模板降级：基于用户输入返回结构化模板，保持交互连贯。
        """
        return (
            "🤖 **Auto-generated stub reply**\n\n"
            "We couldn't reach the full AI service right now, but here's a quick overview "
            "of what you asked:\n\n"
            f"> {user_message.strip()}\n\n"
            "— *This is a provisional response. Please check back shortly for a detailed answer.*"
        )

    async def _cached_fallback(self, user_message: str) -> str:
        """
        缓存降级：若命中缓存则返回历史回答，否则退化为 simple_fallback。
        """
        cached = self._cache_get(self._cache_key(user_message))
        if cached:
            return f"💾 *Cached answer:* {cached}"
        # miss → simple
        return await self._simple_fallback(user_message)

    # ------------------- Primary Assistant ------------------- #

    async def _execute_primary_assistant(
        self, assistant_id: str, user_message: str
    ) -> str:
        """
        调用 Assistant API（Beta）并等待 run 完成。
        """
        thread = await self.primary.beta.threads.create(
            messages=[{"role": "user", "content": user_message}]
        )
        run = await self.primary.beta.threads.runs.create(
            thread_id=thread.id, assistant_id=assistant_id
        )
        # 轮询等待，生产可使用 webhook
        run = await self._wait_for_run(self.primary, thread.id, run.id)
        messages = await self.primary.beta.threads.messages.list(thread_id=thread.id)
        # 返回最后一条 assistant 角色消息
        for msg in reversed(messages.data):
            if msg.role == "assistant":
                return msg.content[0].text.value
        raise RuntimeError("Assistant produced no response")

    # ------------------- Fallback Assistant ------------------- #

    async def _execute_fallback_assistant(self, user_message: str) -> str:
        """
        备用途径：直接使用 Chat Completion（如 gpt-3.5-turbo）。
        """
        completion = await self.fallback.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": user_message}],
            timeout=20,
        )
        return completion.choices[0].message.content

    # ---------------------- Helpers ---------------------- #

    async def _wait_for_run(
        self, client: OpenAI, thread_id: str, run_id: str, poll_interval: float = 1.5
    ):
        """
        轮询 run 状态直到完成或报错。
        """
        while True:
            run = await client.beta.threads.runs.retrieve(
                thread_id=thread_id, run_id=run_id
            )
            if run.status in ("completed", "failed", "cancelled", "expired"):
                return run
            await asyncio.sleep(poll_interval)

    def _cache_key(self, user_message: str) -> str:
        """
        稳定哈希生成缓存键（确保大小写与空白差异不影响）。
        """
        normalized = " ".join(user_message.strip().lower().split())
        return hashlib.sha256(normalized.encode()).hexdigest()

    def _cache_get(self, key: str) -> Optional[str]:
        return self._local_cache.get(key)

    def _cache_set(self, key: str, value: str) -> None:
        # 缓存最长保存 6 h（示例逻辑）
        self._local_cache[key] = value
        # 生产环境可加过期时间，或使用外部缓存