Robust Ollama API Usage: Error Handling, Monitoring, Utilities, and Docker Deployment

4. Error Handling and Monitoring

The article defines an ollamaService class that encapsulates the Ollama client with configurable host, retry attempts, and retry delay. A retry_on_failure decorator retries a wrapped function up to the specified number of attempts, logging each failure and applying exponential back‑off. The safe_generate method calls self.client.generate, records start and end timestamps, logs duration and token count, and returns a dictionary containing success, data, and metrics. It catches ollama.ResponseError, ConnectionError, and generic Exception, logging detailed error messages and returning structured error information.

A health_check method attempts to list available models and perform a simple generation request (model "llama3", prompt "test") to verify API responsiveness, returning status, model count, and a flag indicating API health.

The monitored_session context manager logs the start and end of a named session, measures its duration, and logs any exceptions that occur within the session.

import logging
from functools import wraps
from contextlib import contextmanager
import ollama
import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class ollamaService:
    """Service class with full error handling"""
    def __init__(self, host: str = "http://localhost:11434", retry_attempts: int = 3, retry_delay: float = 1.0):
        self.client = ollama.Client(host=host)
        self.retry_attempts = retry_attempts
        self.retry_delay = retry_delay

    def retry_on_failure(self, func):
        """Retry decorator"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(self.retry_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    logger.warning(f"尝试 {attempt + 1}/{self.retry_attempts} 失败: {str(e)}")
                    if attempt < self.retry_attempts - 1:
                        time.sleep(self.retry_delay * (attempt + 1))
            logger.error(f"所有重试尝试均失败: {str(last_exception)}")
            raise last_exception
        return wrapper

    @retry_on_failure
    def safe_generate(self, model: str, prompt: str, **kwargs) -> Dict:
        """Safe generate method with error handling"""
        try:
            start_time = time.time()
            response = self.client.generate(model=model, prompt=prompt, **kwargs)
            end_time = time.time()
            duration = end_time - start_time
            logger.info(f"生成完成 - 模型: {model}, 耗时: {duration:.2f}s, token数: {len(response.get('response', ''))}")
            return {"success": True, "data": response, "metrics": {"duration": duration, "model": model}}
        except ollama.ResponseError as e:
            logger.error(f"API响应错误: {e.error}")
            return {"success": False, "error": f"API错误: {e.error}", "error_type": "api_error"}
        except ConnectionError as e:
            logger.error(f"连接错误: {str(e)}")
            return {"success": False, "error": f"连接失败: {str(e)}", "error_type": "connection_error"}
        except Exception as e:
            logger.error(f"未知错误: {str(e)}", exc_info=True)
            return {"success": False, "error": f"内部错误: {str(e)}", "error_type": "unknown_error"}

    def health_check(self) -> Dict:
        """Health check"""
        try:
            models = self.client.list()
            test_response = self.client.generate(model="llama3", prompt="test", options={"num_predict": 5})
            return {"status": "healthy", "available_models": len(models.get("models", [])), "api_responding": True}
        except Exception as e:
            logger.error(f"健康检查失败: {str(e)}")
            return {"status": "unhealthy", "error": str(e), "available_models": 0, "api_responding": False}

    @contextmanager
    def monitored_session(self, session_name: str):
        """Monitored session context manager"""
        logger.info(f"开始会话: {session_name}")
        session_start = time.time()
        try:
            yield
        except Exception as e:
            logger.error(f"会话 {session_name} 异常: {str(e)}")
            raise
        finally:
            session_duration = time.time() - session_start
            logger.info(f"结束会话: {session_name}, 持续时间: {session_duration:.2f}秒")

5. Practical Utility Functions

The utils/ollama_utils.py module provides helper functions for common tasks: extract_code_blocks uses a regular expression to find markdown‑style code fences (e.g., ```python) and returns a list of dictionaries with language and code. cache_response creates a cache directory, hashes the prompt to generate a filename, and stores the prompt, response, and timestamp as JSON. load_cached_response reads the cached file, checks whether it is younger than max_age_hours, and returns the cached response or None. format_prompt_for_model tailors the prompt based on the model type: for code‑oriented models it adds a request for runnable code with comments; for instruct models it adds structural and accuracy requirements; otherwise it returns the original prompt. integrate_ollama_in_project shows how to import the utilities and instantiate an Ollama client. get_intelligent_response demonstrates a full workflow: check cache, format the prompt, call client.chat_completion, cache the result, and return the response.

# utils/ollama_utils.py
"""ollama 实用工具函数"""
import re
import hashlib
from pathlib import Path
import json
import time

def extract_code_blocks(text: str) -> List[Dict]:
    """从文本中提取代码块
    返回格式:
    [
        {"language": "python", "code": "print('hello')"},
        ...
    ]
    """
    code_blocks = []
    pattern = r'```(\w+)
(.*?)
```'
    for match in re.finditer(pattern, text, re.DOTALL):
        language = match.group(1)
        code = match.group(2).strip()
        code_blocks.append({"language": language, "code": code})
    return code_blocks

def cache_response(prompt: str, response: str, cache_dir: str = "./ollama_cache") -> str:
    """缓存API响应"""
    Path(cache_dir).mkdir(parents=True, exist_ok=True)
    prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
    cache_file = Path(cache_dir) / f"{prompt_hash}.json"
    cache_data = {"prompt": prompt, "response": response, "timestamp": time.time()}
    with open(cache_file, 'w', encoding='utf-8') as f:
        json.dump(cache_data, f, ensure_ascii=False, indent=2)
    return str(cache_file)

def load_cached_response(prompt: str, cache_dir: str = "./ollama_cache", max_age_hours: float = 24.0) -> Optional[str]:
    """加载缓存的响应"""
    prompt_hash = hashlib.md5(prompt.encode()).hexdigest()
    cache_file = Path(cache_dir) / f"{prompt_hash}.json"
    if not cache_file.exists():
        return None
    try:
        with open(cache_file, 'r', encoding='utf-8') as f:
            cache_data = json.load(f)
        cache_age = time.time() - cache_data.get("timestamp", 0)
        if cache_age > max_age_hours * 3600:
            return None
        return cache_data.get("response")
    except (json.JSONDecodeError, KeyError, IOError):
        return None

def format_prompt_for_model(model_type: str, prompt: str) -> str:
    if "code" in model_type.lower():
        return f"""请用清晰、可运行的代码回答以下问题：
{prompt}
要求：
1. 提供完整的代码示例
2. 添加必要的注释
3. 解释关键实现细节
"""
    elif "instruct" in model_type.lower():
        return f"""请遵循以下指令：
{prompt}
请确保回答：
1. 结构化
2. 准确
3. 实用
"""
    else:
        return prompt

def integrate_ollama_in_project():
    """在项目中使用ollama的完整示例"""
    from utils.ollama_utils import (extract_code_blocks, cache_response, load_cached_response, format_prompt_for_model)
    client = ollamaPythonClient()

def get_intelligent_response(prompt: str, model: str = "llama3") -> str:
    """获取智能响应（带缓存）"""
    cached = load_cached_response(prompt)
    if cached:
        print("使用缓存响应")
        return cached
    formatted_prompt = format_prompt_for_model(model, prompt)
    print("调用ollama API...")
    result = client.chat_completion(model=model, messages=[{"role": "user", "content": formatted_prompt}])
    response = result['message']['content']
    cache_response(prompt, response)
    return response

6. Deployment and Configuration Recommendations

The article supplies a sample config/_config.yaml that sets the host URL, timeout, retry policy, model aliases (default, coding, creative), cache options (enabled, directory, max age), and monitoring settings (enabled, log level, metrics directory).

# config/_config.yaml
host: "http://localhost:11434"
timeout: 120
retry_attempts: 3
retry_delay: 2.0
models:
  default: "llama3"
  coding: "codellama"
  creative: "mistral"
cache:
  enabled: true
  directory: "./cache/"
  max_age_hours: 24
monitoring:
  enabled: true
  log_level: "INFO"
  metrics_dir: "./metrics"

A minimal Dockerfile is provided to containerize the service. It starts from python:3.9-slim, installs curl, runs the Ollama install script, pulls the llama3 and codellama models, installs Python dependencies from requirements.txt, copies the application code, exposes port 11434, and finally starts both the Ollama server and the Python application.

# Dockerfile
FROM python:3.9-slim
WORKDIR /app
# Install Ollama
RUN apt-get update && apt-get install -y curl
RUN curl -fsSL https://ollama.com/install.sh | sh
# Pull models
RUN ollama pull llama3
RUN ollama pull codellama
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Expose Ollama port
EXPOSE 11434
# Start Ollama server and Python app
CMD ["sh", "-c", "ollama serve & python app/main.py"]

Key takeaways include always verifying model availability before making requests, using streaming output for long texts to avoid blocking, tuning timeout values according to task complexity, implementing a cache layer to reduce duplicate API calls, and adding monitoring and logging for easier debugging and performance analysis.