Advanced Guide - Ollama Function Calling with Python
1. Introduction to Ollama Function Calling
Ollama now supports function calling capabilities, allowing you to pass Python functions as tools to the model. This enables powerful integrations between your Python code and Ollama's language models.
Key Features:
- Full typing support
- Python function integration
- Automated JSON schema generation
- Support for existing Python libraries
- Enhanced error handling
2. Prerequisites
Before starting, ensure you have:
-
Python: Version 3.8 or newer
python --version
-
Ollama: Latest version installed
pip install -U ollama
-
Optional Dependencies:
pip install pydantic requests typing_extensions
3. Basic Function Calling
Step 1: Define Your Function
Create a well-documented Python function with type hints:
def calculate_area(length: float, width: float) -> float:
"""
Calculate the area of a rectangle.
Args:
length: The length of the rectangle
width: The width of the rectangle
Returns:
float: The area of the rectangle
"""
return length * width
Step 2: Use Function as Tool
Pass the function to Ollama:
import ollama
response = ollama.chat(
'llama2',
messages=[{
'role': 'user',
'content': 'What is the area of a rectangle with length 5 and width 3?'
}],
tools=[calculate_area]
)
Step 3: Handle Function Calls
Process the model's response and execute functions:
def handle_tool_calls(response, available_functions):
"""Process tool calls from model response"""
for tool in response.message.tool_calls or []:
function_name = tool.function.name
function_to_call = available_functions.get(function_name)
if function_to_call:
try:
result = function_to_call(**tool.function.arguments)
print(f"Function {function_name} output:", result)
except Exception as e:
print(f"Error executing {function_name}:", str(e))
else:
print(f"Function not found: {function_name}")
# Usage
available_functions = {
'calculate_area': calculate_area
}
handle_tool_calls(response, available_functions)
4. Advanced Function Integration
Multiple Function Tools
def get_weather(city: str) -> dict:
"""Get weather information for a city"""
return {"temperature": 20, "conditions": "sunny"}
def convert_temperature(celsius: float) -> float:
"""Convert Celsius to Fahrenheit"""
return (celsius * 9/5) + 32
# Pass multiple functions
response = ollama.chat(
'llama2',
messages=[{
'role': 'user',
'content': 'What is the temperature in London in Fahrenheit?'
}],
tools=[get_weather, convert_temperature]
)
Working with External APIs
import requests
def fetch_api_data(url: str, method: str = "GET") -> dict:
"""
Fetch data from an external API.
Args:
url: The API endpoint URL
method: HTTP method (default: GET)
Returns:
dict: API response data
"""
response = requests.request(method=method, url=url)
return response.json()
# Use in Ollama
response = ollama.chat(
'llama2',
messages=[{
'role': 'user',
'content': 'Get data from https://api.example.com/data'
}],
tools=[fetch_api_data]
)
5. Error Handling and Validation
Input Validation
from typing import Optional
from pydantic import BaseModel, Field
class CalculationInput(BaseModel):
value: float = Field(..., description="Input value for calculation")
precision: Optional[int] = Field(2, description="Decimal precision")
def process_calculation(input_data: CalculationInput) -> float:
"""
Process calculation with validated input.
Args:
input_data: Validated calculation input
Returns:
float: Processed result
"""
return round(input_data.value * 1.5, input_data.precision)
Error Handling Wrapper
from functools import wraps
from typing import Callable, Any
def safe_function_call(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs) -> Any:
try:
return func(*args, **kwargs)
except Exception as e:
return {
"error": str(e),
"function": func.__name__,
"args": args,
"kwargs": kwargs
}
return wrapper
@safe_function_call
def divide_numbers(a: float, b: float) -> float:
"""Safely divide two numbers"""
return a / b
6. Working with Complex Types
Custom Object Types
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class Point:
x: float
y: float
def calculate_distance(point1: Point, point2: Point) -> float:
"""
Calculate distance between two points.
Args:
point1: First point coordinates
point2: Second point coordinates
Returns:
float: Distance between points
"""
return ((point2.x - point1.x) ** 2 + (point2.y - point1.y) ** 2) ** 0.5
Working with Collections
def process_data_batch(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Process a batch of data items.
Args:
items: List of data items to process
Returns:
List[Dict[str, Any]]: Processed items
"""
return [{"processed": item} for item in items]
7. Best Practices
1. Function Documentation
Always provide clear docstrings:
def analyze_text(
text: str,
max_length: Optional[int] = None,
language: str = "en"
) -> Dict[str, Any]:
"""
Analyze text content for various metrics.
Args:
text: Input text to analyze
max_length: Maximum text length to process
language: Language code (default: en)
Returns:
Dict[str, Any]: Analysis results including:
- word_count: Number of words
- sentiment: Sentiment score
- language: Detected language
"""
# Implementation
pass
2. Type Hints
Use comprehensive type hints:
from typing import Union, Optional, Literal
def process_value(
value: Union[int, float],
operation: Literal["add", "multiply"] = "add",
factor: Optional[float] = None
) -> float:
"""Process a numeric value"""
if operation == "add":
return value + (factor or 0)
return value * (factor or 1)
3. Modular Design
Organize related functions:
class DataProcessor:
@staticmethod
def clean_text(text: str) -> str:
"""Clean input text"""
return text.strip()
@staticmethod
def validate_input(data: Dict[str, Any]) -> bool:
"""Validate input data"""
return all(required in data for required in ["id", "value"])
8. Advanced Usage Patterns
Async Function Support
import asyncio
from typing import Awaitable
async def fetch_data(url: str) -> dict:
"""
Asynchronously fetch data.
Args:
url: Data source URL
Returns:
dict: Fetched data
"""
# Async implementation
pass
# Handle async functions
async def process_async_tool_calls(
response,
available_functions: Dict[str, Awaitable]
):
"""Process async tool calls"""
for tool in response.message.tool_calls or []:
func = available_functions.get(tool.function.name)
if func:
result = await func(**tool.function.arguments)
print(f"Async result: {result}")
Function Chaining
def chain_functions(*functions):
"""
Chain multiple functions together.
Args:
*functions: Functions to chain
Returns:
Callable: Chained function
"""
def chained(*args, **kwargs):
result = functions[0](*args, **kwargs)
for func in functions[1:]:
result = func(result)
return result
return chained
# Usage
process_pipeline = chain_functions(
fetch_data,
clean_data,
analyze_data
)
9. Testing and Debugging
Unit Testing Functions
import unittest
class TestCalculations(unittest.TestCase):
def test_calculate_area(self):
"""Test area calculation"""
self.assertEqual(calculate_area(2, 3), 6)
self.assertEqual(calculate_area(0, 5), 0)
if __name__ == '__main__':
unittest.main()
Debugging Tools
def debug_function_call(func, *args, **kwargs):
"""
Debug function execution.
Args:
func: Function to debug
*args: Positional arguments
**kwargs: Keyword arguments
"""
print(f"Calling {func.__name__}")
print(f"Args: {args}")
print(f"Kwargs: {kwargs}")
result = func(*args, **kwargs)
print(f"Result: {result}")
return result
10. Performance Optimization
Caching Results
from functools import lru_cache
@lru_cache(maxsize=100)
def expensive_calculation(x: int) -> int:
"""
Cached expensive calculation.
Args:
x: Input value
Returns:
int: Calculated result
"""
# Expensive computation
return x ** 2
Batch Processing
def process_batch(items: List[Any], batch_size: int = 100):
"""
Process items in batches.
Args:
items: Items to process
batch_size: Size of each batch
"""
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
# Process batch
yield batch
11. Security Considerations
Input Sanitization
import re
from typing import Optional
def sanitize_input(text: str) -> Optional[str]:
"""
Sanitize user input.
Args:
text: Input text to sanitize
Returns:
Optional[str]: Sanitized text or None if invalid
"""
# Remove potentially dangerous characters
sanitized = re.sub(r'[^\w\s-]', '', text)
return sanitized if sanitized else None
Function Access Control
def restrict_function_access(func, allowed_roles: List[str]):
"""
Restrict function access by role.
Args:
func: Function to protect
allowed_roles: List of allowed roles
"""
def wrapper(*args, **kwargs):
# Check role access
if current_role not in allowed_roles:
raise PermissionError("Access denied")
return func(*args, **kwargs)
return wrapper
12. Recap:
Function calling in Ollama provides a powerful way to integrate Python functions with language models. By following best practices and implementing proper error handling, you can create robust and maintainable applications that leverage both Python's capabilities and Ollama's language understanding.
Remember to:
- Use type hints and proper documentation
- Implement comprehensive error handling
- Follow security best practices
- Test thoroughly
- Optimize for performance when needed
For more information and updates, visit the Ollama documentation.