INTRODUCTION TO OLLAMA AND ITS REVOLUTIONARY APPROACH
Imagine having the power of large language models running directly on your own hardware, without dependency on cloud services, without usage limits, and without privacy concerns about your data being sent to external servers. This is exactly what Ollama brings to the table. Ollama is an open-source project that makes it incredibly simple to run large language models locally on your machine. It handles all the complexity of model management, GPU acceleration, and provides a clean, straightforward API that developers can integrate into their applications.
The beauty of Ollama lies in its simplicity. While other solutions require you to understand complex model architectures, manage CUDA installations, or deal with intricate configuration files, Ollama abstracts all of this away. You simply download the Ollama application, pull the models you want to use, and start making API calls. The models run entirely on your local machine, which means your data never leaves your infrastructure. This is particularly important for enterprises dealing with sensitive information or developers who want to prototype without incurring cloud API costs.
The Ollama API follows RESTful principles and provides endpoints for generating text completions, streaming responses, managing models, and even generating embeddings for semantic search applications. Whether you are building a chatbot, a code assistant, a content generation tool, or any application that needs natural language understanding and generation, the Ollama API provides the foundation you need.
UNDERSTANDING THE ARCHITECTURE AND CORE CONCEPTS
Before we dive into building applications, it is essential to understand how Ollama works under the hood and what makes it different from other LLM solutions. Ollama runs as a local server on your machine, typically listening on port 11434. When you start Ollama, it launches a background service that manages model loading, inference, and resource allocation. This service is what your applications will communicate with through HTTP requests.
The models themselves are stored in a local repository on your machine. When you pull a model using Ollama, it downloads the model weights and configuration files to your local storage. These models are optimized versions of popular open-source language models like Llama, Mistral, CodeLlama, and many others. Ollama uses quantization techniques to reduce the model sizes while maintaining performance, which means you can run powerful models even on consumer-grade hardware.
One of the most important concepts to understand is the difference between completion and chat endpoints. The completion endpoint is designed for single-turn text generation where you provide a prompt and receive a completion. The chat endpoint, on the other hand, maintains conversation context and is designed for multi-turn dialogues. The chat endpoint accepts an array of messages with roles like system, user, and assistant, allowing you to build conversational applications that remember context across multiple exchanges.
Another crucial concept is streaming versus non-streaming responses. When you make a non-streaming request, your application waits until the entire response is generated before receiving it. This can lead to perceived latency, especially for longer responses. Streaming responses, however, send the generated text in chunks as it is being produced, allowing you to display partial results to users immediately. This creates a much more responsive user experience, similar to how ChatGPT displays responses word by word.
SETTING UP YOUR DEVELOPMENT ENVIRONMENT
The first step in your journey with Ollama is getting it installed and running on your development machine. The installation process varies slightly depending on your operating system, but Ollama provides native installers for macOS, Linux, and Windows. For macOS users, you can download the Ollama application from the official website and drag it to your Applications folder. Linux users can install it using a simple curl command that downloads and runs the installation script. Windows users can download the Windows installer and run it like any other application.
Once Ollama is installed, you need to start the Ollama service. On macOS and Windows, simply launching the Ollama application starts the background service automatically. On Linux, the installation script typically sets up Ollama as a systemd service that starts automatically. You can verify that Ollama is running by opening a terminal and executing the command to list available models. If Ollama is running correctly, you will see a list of models, which might be empty if you have not pulled any models yet.
The next step is to pull a model that you will use for development. For this tutorial, we will use the Llama 2 model, which is a powerful general-purpose language model. To pull this model, you open a terminal and execute the pull command with the model name. The download might take some time depending on your internet connection, as these models can be several gigabytes in size. Once the download completes, Ollama automatically makes the model available for use.
To verify that everything is working correctly, you can test the model using the command line interface. Simply run the Ollama run command followed by the model name, and you will enter an interactive chat session. Type a message and press enter to see the model generate a response. This confirms that Ollama is properly installed, the model is loaded, and the inference engine is working. Once you see a successful response, you are ready to start building applications that use the API programmatically.
MAKING YOUR FIRST API REQUEST
Now that Ollama is running and you have a model available, let us make our first API request using Python. We will start with a simple example that demonstrates the basic structure of an Ollama API call. The Ollama API is a REST API, which means we can interact with it using standard HTTP requests. Python's requests library makes this incredibly straightforward.
Here is a simple example that sends a prompt to Ollama and receives a completion:
import requests
import json
# Define the API endpoint
url = "http://localhost:11434/api/generate"
# Prepare the request payload
payload = {
"model": "llama2",
"prompt": "Explain quantum computing in simple terms",
"stream": False
}
# Make the POST request
response = requests.post(url, json=payload)
# Parse and print the response
if response.status_code == 200:
result = response.json()
print(result['response'])
else:
print(f"Error: {response.status_code}")
Let us break down what is happening in this code. First, we import the necessary libraries. The requests library handles HTTP communication, and json helps us work with JSON data structures. We then define the URL for the Ollama API endpoint. Since Ollama runs locally, we use localhost with port 11434, which is Ollama's default port. The endpoint path is /api/generate, which is the completion endpoint.
The payload dictionary contains the parameters for our request. The model parameter specifies which model we want to use for generation. In this case, we are using llama2, which should match the name of a model you have pulled. The prompt parameter contains the text we want the model to complete or respond to. The stream parameter is set to False, which means we want to receive the entire response at once rather than as a streaming response.
We then make a POST request to the API endpoint, passing our payload as JSON. The requests library automatically serializes our Python dictionary to JSON format. After making the request, we check the status code to ensure the request was successful. A status code of 200 indicates success. We then parse the JSON response and extract the generated text from the response field.
This basic pattern forms the foundation of all interactions with the Ollama API. You send a POST request with a JSON payload containing your parameters, and you receive a JSON response containing the generated text and metadata. Understanding this fundamental request-response cycle is crucial before moving on to more advanced features.
IMPLEMENTING STREAMING RESPONSES FOR BETTER USER EXPERIENCE
While the previous example works perfectly fine, it has a significant limitation. For longer responses, your application will appear frozen while waiting for the complete generation to finish. This is where streaming responses become invaluable. Streaming allows you to receive and display partial results as they are generated, creating a much more responsive and engaging user experience.
Implementing streaming with the Ollama API requires a slightly different approach because the response is not a single JSON object but rather a stream of JSON objects, one for each generated token. Here is how you implement streaming:
import requests
import json
# Define the API endpoint
url = "http://localhost:11434/api/generate"
# Prepare the request payload with streaming enabled
payload = {
"model": "llama2",
"prompt": "Write a short story about a robot learning to paint",
"stream": True
}
# Make the POST request with streaming enabled
response = requests.post(url, json=payload, stream=True)
# Process the streaming response
if response.status_code == 200:
full_response = ""
for line in response.iter_lines():
if line:
# Parse each line as JSON
chunk = json.loads(line)
# Extract the response fragment
if 'response' in chunk:
text_fragment = chunk['response']
full_response += text_fragment
# Print each fragment as it arrives
print(text_fragment, end='', flush=True)
# Check if generation is complete
if chunk.get('done', False):
print("\n\nGeneration complete!")
break
print(f"\n\nFull response length: {len(full_response)} characters")
else:
print(f"Error: {response.status_code}")
The key difference here is that we set the stream parameter to True in our payload, and we also add stream=True to the requests.post call. This tells the requests library to not download the entire response at once but to keep the connection open and allow us to read the response incrementally.
We then iterate over the response using iter_lines, which yields each line of the streaming response. Each line contains a JSON object representing one chunk of the generated text. We parse each line using json.loads and extract the response field, which contains the text fragment for that particular token or group of tokens.
As we receive each fragment, we can immediately display it to the user using print with end='' to avoid adding newlines, and flush=True to ensure the output appears immediately rather than being buffered. We also accumulate the fragments in a full_response variable so we have the complete text when generation finishes.
The streaming response includes a done field that becomes True when the generation is complete. This signals that we have received all fragments and can stop processing the stream. The final chunk also includes additional metadata like the total duration, the number of tokens generated, and performance metrics.
Streaming is particularly important for applications with user interfaces where you want to show progress and keep users engaged. It transforms the experience from waiting for a potentially long operation to seeing results appear in real-time, similar to how modern AI assistants work.
BUILDING CONVERSATIONAL APPLICATIONS WITH THE CHAT ENDPOINT
While the generate endpoint is useful for single-turn completions, most modern LLM applications are conversational in nature. Users expect to have multi-turn dialogues where the model remembers previous exchanges and maintains context. This is where the chat endpoint becomes essential. The chat endpoint is specifically designed for conversational interactions and handles context management more elegantly than manually concatenating prompts.
The chat endpoint accepts an array of message objects, where each message has a role and content. The role can be system, user, or assistant. System messages set the behavior and personality of the assistant. User messages represent what the user says. Assistant messages represent previous responses from the model. By maintaining this message history, the model can provide contextually relevant responses.
Here is an example of using the chat endpoint for a multi-turn conversation:
import requests
import json
# Define the chat API endpoint
url = "http://localhost:11434/api/chat"
# Initialize conversation history
conversation_history = [
{
"role": "system",
"content": "You are a helpful programming tutor who explains concepts clearly and provides practical examples."
}
]
def send_message(user_message):
"""Send a message and get a response while maintaining conversation history."""
# Add user message to history
conversation_history.append({
"role": "user",
"content": user_message
})
# Prepare the request payload
payload = {
"model": "llama2",
"messages": conversation_history,
"stream": False
}
# Make the request
response = requests.post(url, json=payload)
if response.status_code == 200:
result = response.json()
assistant_message = result['message']['content']
# Add assistant response to history
conversation_history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
else:
return f"Error: {response.status_code}"
# Example conversation
print("User: What is a Python decorator?")
response1 = send_message("What is a Python decorator?")
print(f"Assistant: {response1}\n")
print("User: Can you show me a simple example?")
response2 = send_message("Can you show me a simple example?")
print(f"Assistant: {response2}\n")
print("User: How would I use that in a real application?")
response3 = send_message("How would I use that in a real application?")
print(f"Assistant: {response3}\n")
This example demonstrates several important concepts. First, we maintain a conversation_history list that stores all messages in the conversation. We initialize this with a system message that defines the assistant's role and behavior. This system message influences how the model responds throughout the entire conversation.
The send_message function encapsulates the logic for sending a user message and receiving a response. It first appends the user's message to the conversation history with the role set to user. It then constructs the API payload with the complete message history. This is crucial because the model needs to see all previous messages to maintain context.
After receiving the response, we extract the assistant's message from the result and append it to the conversation history with the role set to assistant. This ensures that the next request will include this response as part of the context. The conversation history grows with each exchange, allowing the model to reference earlier parts of the conversation.
Notice how the second and third questions reference the previous context. The user asks "Can you show me a simple example?" without specifying what they want an example of. The model understands from the conversation history that they are asking about Python decorators. Similarly, the third question uses "that" to refer to the decorator example, and the model correctly interprets this reference.
The system message is particularly powerful for customizing the assistant's behavior. You can use it to define the assistant's personality, expertise, response style, or any constraints you want to impose. For example, you could create a system message that makes the assistant respond in a specific format, focus on certain topics, or adopt a particular tone.
ADVANCED PARAMETER TUNING FOR OPTIMAL RESULTS
The Ollama API provides numerous parameters that allow you to fine-tune the generation process to achieve the exact behavior you need for your application. Understanding these parameters and how they affect the output is crucial for building production-quality applications. Let us explore the most important parameters and how to use them effectively.
The temperature parameter controls the randomness of the generation. A temperature of 0 makes the model completely deterministic, always choosing the most likely next token. Higher temperatures introduce more randomness, making the output more creative but potentially less coherent. For factual question-answering, you typically want a low temperature around 0.1 to 0.3. For creative writing or brainstorming, you might use temperatures between 0.7 and 1.0.
Here is an example demonstrating different temperature settings:
import requests
import json
url = "http://localhost:11434/api/generate"
def generate_with_temperature(prompt, temperature):
"""Generate text with a specific temperature setting."""
payload = {
"model": "llama2",
"prompt": prompt,
"stream": False,
"options": {
"temperature": temperature
}
}
response = requests.post(url, json=payload)
if response.status_code == 200:
return response.json()['response']
return None
# Test the same prompt with different temperatures
prompt = "Complete this sentence: The future of artificial intelligence is"
print("Temperature 0.1 (Very Deterministic):")
print(generate_with_temperature(prompt, 0.1))
print("\n")
print("Temperature 0.7 (Balanced):")
print(generate_with_temperature(prompt, 0.7))
print("\n")
print("Temperature 1.5 (Very Creative):")
print(generate_with_temperature(prompt, 1.5))
print("\n")
The top_p parameter, also known as nucleus sampling, provides another way to control randomness. Instead of selecting from all possible tokens weighted by probability, top_p considers only the smallest set of tokens whose cumulative probability exceeds the threshold. A top_p value of 0.9 means the model considers only the top tokens that together account for 90 percent of the probability mass. This often produces more coherent results than temperature alone.
The top_k parameter limits the model to considering only the k most likely tokens at each step. For example, top_k of 40 means the model only considers the 40 most probable next tokens. This can help prevent the model from occasionally selecting very unlikely tokens that might derail the generation.
The num_predict parameter controls the maximum number of tokens to generate. This is useful when you want to limit response length. However, be aware that the model might stop generating before reaching this limit if it produces an end-of-sequence token.
Here is a more comprehensive example showing multiple parameters working together:
import requests
import json
url = "http://localhost:11434/api/generate"
def generate_with_options(prompt, temperature=0.7, top_p=0.9, top_k=40, num_predict=200):
"""Generate text with comprehensive parameter control."""
payload = {
"model": "llama2",
"prompt": prompt,
"stream": False,
"options": {
"temperature": temperature,
"top_p": top_p,
"top_k": top_k,
"num_predict": num_predict,
"repeat_penalty": 1.1,
"stop": ["\n\n", "END"]
}
}
response = requests.post(url, json=payload)
if response.status_code == 200:
result = response.json()
return {
"text": result['response'],
"tokens_generated": result.get('eval_count', 0),
"generation_time": result.get('eval_duration', 0) / 1e9 # Convert to seconds
}
return None
# Generate a product description with controlled parameters
prompt = "Write a compelling product description for a smart water bottle that tracks hydration:"
result = generate_with_options(
prompt=prompt,
temperature=0.8, # Slightly creative for marketing copy
top_p=0.95, # Allow diverse word choices
top_k=50, # Moderate vocabulary restriction
num_predict=150 # Limit to reasonable description length
)
if result:
print(f"Generated Description:\n{result['text']}\n")
print(f"Tokens Generated: {result['tokens_generated']}")
print(f"Generation Time: {result['generation_time']:.2f} seconds")
The repeat_penalty parameter helps prevent the model from repeating the same phrases or falling into repetitive patterns. A value of 1.0 means no penalty, while higher values increasingly discourage repetition. Values between 1.1 and 1.3 typically work well.
The stop parameter allows you to specify sequences that should terminate generation. When the model generates any of these sequences, it immediately stops. This is useful for creating structured outputs or ensuring the model does not continue beyond a certain point. In the example above, we stop at double newlines or the word "END", which could be useful for generating distinct paragraphs or sections.
IMPLEMENTING ERROR HANDLING AND RETRY LOGIC
Production applications need robust error handling to deal with various failure scenarios. The Ollama API can fail for several reasons including the service being down, the model not being available, network issues, or resource constraints. Implementing proper error handling and retry logic ensures your application gracefully handles these situations.
Here is a comprehensive error handling implementation:
import requests
import json
import time
from typing import Optional, Dict, Any
class OllamaAPIError(Exception):
"""Custom exception for Ollama API errors."""
pass
class OllamaClient:
"""A robust client for interacting with the Ollama API."""
def __init__(self, base_url="http://localhost:11434", max_retries=3, retry_delay=1.0):
"""
Initialize the Ollama client.
Args:
base_url: The base URL for the Ollama API
max_retries: Maximum number of retry attempts for failed requests
retry_delay: Initial delay between retries in seconds (uses exponential backoff)
"""
self.base_url = base_url
self.max_retries = max_retries
self.retry_delay = retry_delay
def _make_request(self, endpoint: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Make a request to the Ollama API with retry logic.
Args:
endpoint: API endpoint path
payload: Request payload
Returns:
Response data as dictionary or None if all retries failed
Raises:
OllamaAPIError: If the request fails after all retries
"""
url = f"{self.base_url}{endpoint}"
last_error = None
for attempt in range(self.max_retries):
try:
response = requests.post(url, json=payload, timeout=60)
# Check for successful response
if response.status_code == 200:
return response.json()
# Handle specific error codes
elif response.status_code == 404:
raise OllamaAPIError(f"Model not found. Please ensure the model is pulled.")
elif response.status_code == 500:
# Server error, might be temporary, retry
last_error = f"Server error (500) on attempt {attempt + 1}"
else:
raise OllamaAPIError(f"API returned status code {response.status_code}: {response.text}")
except requests.exceptions.ConnectionError as e:
last_error = f"Connection error on attempt {attempt + 1}: {str(e)}"
except requests.exceptions.Timeout as e:
last_error = f"Request timeout on attempt {attempt + 1}: {str(e)}"
except requests.exceptions.RequestException as e:
last_error = f"Request exception on attempt {attempt + 1}: {str(e)}"
# If we have more retries, wait before trying again
if attempt < self.max_retries - 1:
wait_time = self.retry_delay * (2 ** attempt) # Exponential backoff
print(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
# All retries exhausted
raise OllamaAPIError(f"Request failed after {self.max_retries} attempts. Last error: {last_error}")
def generate(self, model: str, prompt: str, stream: bool = False, **options) -> Optional[str]:
"""
Generate text using the Ollama API.
Args:
model: Model name to use
prompt: Input prompt
stream: Whether to stream the response
**options: Additional generation options
Returns:
Generated text or None if generation failed
"""
payload = {
"model": model,
"prompt": prompt,
"stream": stream
}
if options:
payload["options"] = options
try:
result = self._make_request("/api/generate", payload)
if result:
return result.get('response', '')
except OllamaAPIError as e:
print(f"Generation failed: {e}")
return None
def check_health(self) -> bool:
"""
Check if the Ollama service is running and accessible.
Returns:
True if service is healthy, False otherwise
"""
try:
response = requests.get(f"{self.base_url}/api/tags", timeout=5)
return response.status_code == 200
except:
return False
# Example usage
if __name__ == "__main__":
client = OllamaClient(max_retries=3, retry_delay=1.0)
# Check if service is available
if not client.check_health():
print("Error: Ollama service is not running or not accessible")
print("Please ensure Ollama is installed and running")
exit(1)
# Generate text with error handling
result = client.generate(
model="llama2",
prompt="Explain the concept of error handling in software development",
temperature=0.7,
num_predict=200
)
if result:
print(f"Generated response:\n{result}")
else:
print("Failed to generate response after multiple attempts")
This implementation demonstrates several important error handling patterns. The OllamaClient class encapsulates all API interactions and provides a clean interface for the rest of your application. The _make_request method implements the retry logic with exponential backoff, which means each retry waits progressively longer before attempting again.
The code handles different types of errors appropriately. Connection errors and timeouts are transient and worth retrying. A 404 status code indicates the model is not available, which is not a transient error and should not be retried. Server errors might be temporary, so we retry those. The timeout parameter on requests prevents the application from hanging indefinitely if the API becomes unresponsive.
The check_health method provides a way to verify the Ollama service is running before attempting to use it. This is particularly useful in containerized environments or when your application starts up, allowing you to provide clear error messages to users if the service is unavailable.
WORKING WITH EMBEDDINGS FOR SEMANTIC SEARCH
Beyond text generation, the Ollama API also provides embedding capabilities. Embeddings are numerical vector representations of text that capture semantic meaning. Text with similar meanings will have similar embeddings, even if the exact words are different. This makes embeddings incredibly useful for semantic search, clustering, classification, and recommendation systems.
The embeddings endpoint takes text as input and returns a high-dimensional vector representing that text. You can then use these vectors to find similar texts, cluster documents, or build recommendation systems. Here is how to work with embeddings:
import requests
import json
import numpy as np
from typing import List, Tuple
class EmbeddingManager:
"""Manages text embeddings using the Ollama API."""
def __init__(self, base_url="http://localhost:11434", model="llama2"):
"""
Initialize the embedding manager.
Args:
base_url: The base URL for the Ollama API
model: The model to use for generating embeddings
"""
self.base_url = base_url
self.model = model
self.embeddings_cache = {}
def get_embedding(self, text: str) -> np.ndarray:
"""
Get the embedding vector for a given text.
Args:
text: Input text to embed
Returns:
Numpy array containing the embedding vector
"""
# Check cache first
if text in self.embeddings_cache:
return self.embeddings_cache[text]
url = f"{self.base_url}/api/embeddings"
payload = {
"model": self.model,
"prompt": text
}
response = requests.post(url, json=payload)
if response.status_code == 200:
result = response.json()
embedding = np.array(result['embedding'])
# Cache the result
self.embeddings_cache[text] = embedding
return embedding
else:
raise Exception(f"Failed to get embedding: {response.status_code}")
def cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
"""
Calculate cosine similarity between two vectors.
Args:
vec1: First vector
vec2: Second vector
Returns:
Cosine similarity score between -1 and 1
"""
dot_product = np.dot(vec1, vec2)
norm1 = np.linalg.norm(vec1)
norm2 = np.linalg.norm(vec2)
return dot_product / (norm1 * norm2)
def find_most_similar(self, query: str, documents: List[str], top_k: int = 3) -> List[Tuple[str, float]]:
"""
Find the most similar documents to a query.
Args:
query: Query text
documents: List of document texts to search
top_k: Number of top results to return
Returns:
List of tuples containing (document, similarity_score)
"""
query_embedding = self.get_embedding(query)
similarities = []
for doc in documents:
doc_embedding = self.get_embedding(doc)
similarity = self.cosine_similarity(query_embedding, doc_embedding)
similarities.append((doc, similarity))
# Sort by similarity score in descending order
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_k]
# Example usage: Building a simple semantic search system
if __name__ == "__main__":
manager = EmbeddingManager(model="llama2")
# Sample document collection
documents = [
"Python is a high-level programming language known for its simplicity and readability.",
"Machine learning is a subset of artificial intelligence that enables systems to learn from data.",
"The Eiffel Tower is a wrought-iron lattice tower located in Paris, France.",
"JavaScript is a programming language commonly used for web development.",
"Deep learning uses neural networks with multiple layers to process complex patterns.",
"The Great Wall of China is an ancient series of fortifications built to protect Chinese states.",
"Java is an object-oriented programming language designed for platform independence.",
"Natural language processing enables computers to understand and generate human language."
]
# Perform semantic search
query = "What programming languages are good for beginners?"
print(f"Query: {query}\n")
print("Most similar documents:")
results = manager.find_most_similar(query, documents, top_k=3)
for i, (doc, score) in enumerate(results, 1):
print(f"\n{i}. Similarity: {score:.4f}")
print(f" Document: {doc}")
This implementation demonstrates how to build a semantic search system using Ollama embeddings. The EmbeddingManager class provides methods for getting embeddings, calculating similarity, and finding the most relevant documents for a query.
The get_embedding method makes a request to the embeddings endpoint and converts the result to a numpy array for easier mathematical operations. We implement caching to avoid repeatedly generating embeddings for the same text, which significantly improves performance when working with a fixed document collection.
The cosine_similarity method calculates how similar two embedding vectors are. Cosine similarity ranges from negative one to positive one, where values closer to one indicate high similarity. This metric is preferred over Euclidean distance for embeddings because it measures the angle between vectors rather than their magnitude, making it more robust to variations in text length.
The find_most_similar method demonstrates a complete semantic search pipeline. It generates an embedding for the query, compares it to all document embeddings, and returns the most similar documents ranked by similarity score. Notice how the search finds programming-related documents even though the query uses different words than the documents.
BUILDING A COMPLETE QUESTION-ANSWERING SYSTEM
Now let us combine everything we have learned to build a complete question-answering system that uses both embeddings for retrieval and generation for answering. This is a common pattern in production applications, often called Retrieval-Augmented Generation or RAG. The system first finds relevant context using semantic search, then uses that context to generate an informed answer.
import requests
import json
import numpy as np
from typing import List, Dict, Optional, Tuple
class DocumentStore:
"""Stores documents and their embeddings for efficient retrieval."""
def __init__(self, base_url="http://localhost:11434", embedding_model="llama2"):
self.base_url = base_url
self.embedding_model = embedding_model
self.documents = []
self.embeddings = []
def add_document(self, text: str, metadata: Optional[Dict] = None):
"""Add a document to the store and generate its embedding."""
# Get embedding for the document
url = f"{self.base_url}/api/embeddings"
payload = {
"model": self.embedding_model,
"prompt": text
}
response = requests.post(url, json=payload)
if response.status_code == 200:
embedding = np.array(response.json()['embedding'])
self.documents.append({
"text": text,
"metadata": metadata or {},
"id": len(self.documents)
})
self.embeddings.append(embedding)
return len(self.documents) - 1
else:
raise Exception(f"Failed to generate embedding: {response.status_code}")
def search(self, query: str, top_k: int = 3) -> List[Dict]:
"""Search for documents similar to the query."""
# Get query embedding
url = f"{self.base_url}/api/embeddings"
payload = {
"model": self.embedding_model,
"prompt": query
}
response = requests.post(url, json=payload)
if response.status_code != 200:
raise Exception(f"Failed to generate query embedding: {response.status_code}")
query_embedding = np.array(response.json()['embedding'])
# Calculate similarities
similarities = []
for i, doc_embedding in enumerate(self.embeddings):
similarity = np.dot(query_embedding, doc_embedding) / (
np.linalg.norm(query_embedding) * np.linalg.norm(doc_embedding)
)
similarities.append((i, similarity))
# Sort and return top results
similarities.sort(key=lambda x: x[1], reverse=True)
results = []
for doc_id, score in similarities[:top_k]:
doc = self.documents[doc_id].copy()
doc['similarity_score'] = score
results.append(doc)
return results
class QuestionAnsweringSystem:
"""A complete QA system using retrieval-augmented generation."""
def __init__(self, base_url="http://localhost:11434", model="llama2"):
self.base_url = base_url
self.model = model
self.document_store = DocumentStore(base_url, model)
def add_knowledge(self, text: str, metadata: Optional[Dict] = None):
"""Add knowledge to the system."""
return self.document_store.add_document(text, metadata)
def answer_question(self, question: str, num_context_docs: int = 3) -> Dict:
"""
Answer a question using retrieval-augmented generation.
Args:
question: The question to answer
num_context_docs: Number of context documents to retrieve
Returns:
Dictionary containing the answer and metadata
"""
# Retrieve relevant context
relevant_docs = self.document_store.search(question, top_k=num_context_docs)
# Build context from retrieved documents
context_parts = []
for i, doc in enumerate(relevant_docs, 1):
context_parts.append(f"Context {i} (relevance: {doc['similarity_score']:.3f}):")
context_parts.append(doc['text'])
context_parts.append("")
context = "\n".join(context_parts)
# Create prompt with context
prompt = f"""Based on the following context, please answer the question. If the context does not contain enough information to answer the question, say so.
{context}
Question: {question}
Answer:"""
# Generate answer
url = f"{self.base_url}/api/generate"
payload = {
"model": self.model,
"prompt": prompt,
"stream": False,
"options": {
"temperature": 0.3, # Lower temperature for factual answers
"num_predict": 300
}
}
response = requests.post(url, json=payload)
if response.status_code == 200:
result = response.json()
return {
"question": question,
"answer": result['response'],
"context_used": relevant_docs,
"num_tokens": result.get('eval_count', 0),
"generation_time": result.get('eval_duration', 0) / 1e9
}
else:
raise Exception(f"Failed to generate answer: {response.status_code}")
# Example usage: Building a knowledge base and answering questions
if __name__ == "__main__":
qa_system = QuestionAnsweringSystem(model="llama2")
# Add knowledge to the system
knowledge_base = [
"Ollama is an open-source tool that allows you to run large language models locally on your computer. It supports various models including Llama 2, Mistral, and CodeLlama.",
"The Ollama API provides endpoints for text generation, chat conversations, and embeddings. It runs as a local server on port 11434 by default.",
"To install Ollama, you can download it from the official website. It is available for macOS, Linux, and Windows. After installation, you can pull models using the ollama pull command.",
"Embeddings are vector representations of text that capture semantic meaning. They are useful for semantic search, clustering, and finding similar documents.",
"The temperature parameter controls randomness in text generation. Lower values make output more deterministic, while higher values increase creativity.",
"Streaming responses allow you to receive generated text incrementally rather than waiting for the complete response. This improves user experience for long generations."
]
print("Adding knowledge to the system...")
for i, text in enumerate(knowledge_base):
qa_system.add_knowledge(text, metadata={"source": f"doc_{i}"})
print("Knowledge base ready!\n")
# Ask questions
questions = [
"How do I install Ollama?",
"What is the temperature parameter used for?",
"Can Ollama run models in the cloud?"
]
for question in questions:
print(f"Question: {question}")
print("-" * 80)
result = qa_system.answer_question(question)
print(f"Answer: {result['answer']}\n")
print(f"Context documents used: {len(result['context_used'])}")
print(f"Generation time: {result['generation_time']:.2f} seconds")
print(f"Tokens generated: {result['num_tokens']}")
print("\n" + "=" * 80 + "\n")
This question-answering system demonstrates a production-ready pattern for building knowledge-based applications. The DocumentStore class manages a collection of documents and their embeddings, providing efficient semantic search capabilities. The QuestionAnsweringSystem class orchestrates the entire process of retrieving relevant context and generating answers.
When a question is asked, the system first searches for the most relevant documents using semantic similarity. It then constructs a prompt that includes these documents as context, along with the original question. This context-aware prompt is sent to the generation endpoint, which produces an answer grounded in the provided knowledge.
The system uses a lower temperature for answer generation to ensure factual, consistent responses. The prompt explicitly instructs the model to acknowledge when it does not have enough information, preventing hallucination. The returned result includes not just the answer but also metadata about which documents were used, how relevant they were, and performance metrics.
This architecture scales well to larger knowledge bases. You can add thousands of documents, and the semantic search will efficiently find the most relevant ones for each question. For even larger collections, you might want to use a dedicated vector database like Chroma or Pinecone, but the pattern remains the same.
MANAGING MODELS PROGRAMMATICALLY
The Ollama API provides endpoints for managing models programmatically, allowing your application to check which models are available, pull new models, and delete models that are no longer needed. This is particularly useful for applications that need to work with multiple models or that need to ensure specific models are available before attempting to use them.
Here is a comprehensive model management implementation:
import requests
import json
from typing import List, Dict, Optional
class ModelManager:
"""Manages Ollama models programmatically."""
def __init__(self, base_url="http://localhost:11434"):
self.base_url = base_url
def list_models(self) -> List[Dict]:
"""
List all models currently available in Ollama.
Returns:
List of dictionaries containing model information
"""
url = f"{self.base_url}/api/tags"
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
result = response.json()
return result.get('models', [])
else:
print(f"Failed to list models: {response.status_code}")
return []
except Exception as e:
print(f"Error listing models: {e}")
return []
def model_exists(self, model_name: str) -> bool:
"""
Check if a specific model is available.
Args:
model_name: Name of the model to check
Returns:
True if model exists, False otherwise
"""
models = self.list_models()
return any(model['name'] == model_name for model in models)
def pull_model(self, model_name: str, stream: bool = True) -> bool:
"""
Pull a model from the Ollama library.
Args:
model_name: Name of the model to pull
stream: Whether to stream the download progress
Returns:
True if successful, False otherwise
"""
url = f"{self.base_url}/api/pull"
payload = {
"name": model_name,
"stream": stream
}
try:
if stream:
response = requests.post(url, json=payload, stream=True)
if response.status_code == 200:
print(f"Pulling model {model_name}...")
for line in response.iter_lines():
if line:
status = json.loads(line)
if 'status' in status:
print(f" {status['status']}")
if status.get('status') == 'success':
print(f"Successfully pulled {model_name}")
return True
return True
else:
print(f"Failed to pull model: {response.status_code}")
return False
else:
response = requests.post(url, json=payload)
return response.status_code == 200
except Exception as e:
print(f"Error pulling model: {e}")
return False
def delete_model(self, model_name: str) -> bool:
"""
Delete a model from local storage.
Args:
model_name: Name of the model to delete
Returns:
True if successful, False otherwise
"""
url = f"{self.base_url}/api/delete"
payload = {"name": model_name}
try:
response = requests.delete(url, json=payload)
if response.status_code == 200:
print(f"Successfully deleted {model_name}")
return True
else:
print(f"Failed to delete model: {response.status_code}")
return False
except Exception as e:
print(f"Error deleting model: {e}")
return False
def get_model_info(self, model_name: str) -> Optional[Dict]:
"""
Get detailed information about a specific model.
Args:
model_name: Name of the model
Returns:
Dictionary with model information or None if not found
"""
models = self.list_models()
for model in models:
if model['name'] == model_name:
return model
return None
def ensure_model_available(self, model_name: str) -> bool:
"""
Ensure a model is available, pulling it if necessary.
Args:
model_name: Name of the model
Returns:
True if model is available or was successfully pulled
"""
if self.model_exists(model_name):
print(f"Model {model_name} is already available")
return True
else:
print(f"Model {model_name} not found, attempting to pull...")
return self.pull_model(model_name)
# Example usage
if __name__ == "__main__":
manager = ModelManager()
# List all available models
print("Available models:")
models = manager.list_models()
for model in models:
print(f" - {model['name']} (size: {model.get('size', 'unknown')})")
print()
# Ensure a specific model is available
required_model = "llama2"
if manager.ensure_model_available(required_model):
print(f"\n{required_model} is ready to use!")
# Get detailed info
info = manager.get_model_info(required_model)
if info:
print(f"\nModel details:")
print(f" Name: {info['name']}")
print(f" Size: {info.get('size', 'unknown')}")
print(f" Modified: {info.get('modified_at', 'unknown')}")
else:
print(f"\nFailed to make {required_model} available")
This model management system provides a complete interface for working with Ollama models programmatically. The list_models method retrieves all currently available models, which is useful for displaying options to users or verifying that required models are present. The model_exists method provides a quick way to check if a specific model is available without parsing the full model list.
The pull_model method handles downloading new models from the Ollama library. When streaming is enabled, it provides progress updates as the model downloads, which is important for user feedback since models can be several gigabytes in size. The delete_model method allows you to remove models that are no longer needed, freeing up disk space.
The ensure_model_available method is particularly useful in production applications. It checks if a model exists and automatically pulls it if not, ensuring your application can always access the models it needs. This is especially valuable in containerized deployments where you might want to pull models on first run rather than baking them into the container image.
FULL PRODUCTION-READY APPLICATION EXAMPLE
Now let us bring everything together into a complete, production-ready application. This application implements a document-based question-answering system with a command-line interface, comprehensive error handling, logging, and configuration management. This represents the kind of code you would actually deploy in a production environment.
#!/usr/bin/env python3
"""
Ollama-powered Document Question-Answering System
A production-ready application that allows users to upload documents,
build a knowledge base, and ask questions that are answered using
retrieval-augmented generation with the Ollama API.
Author: AI Assistant
License: MIT
"""
import requests
import json
import numpy as np
import logging
import sys
import time
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from pathlib import Path
import argparse
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('qa_system.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
@dataclass
class Config:
"""Application configuration."""
ollama_base_url: str = "http://localhost:11434"
default_model: str = "llama2"
embedding_model: str = "llama2"
max_retries: int = 3
retry_delay: float = 1.0
request_timeout: int = 60
temperature: float = 0.3
num_context_docs: int = 3
max_tokens: int = 500
class OllamaAPIException(Exception):
"""Custom exception for Ollama API errors."""
pass
class OllamaClient:
"""Robust client for interacting with the Ollama API."""
def __init__(self, config: Config):
"""
Initialize the Ollama client.
Args:
config: Application configuration object
"""
self.config = config
self.base_url = config.ollama_base_url
logger.info(f"Initialized Ollama client with base URL: {self.base_url}")
def _make_request(self, endpoint: str, payload: Dict, stream: bool = False) -> Optional[Dict]:
"""
Make a request to the Ollama API with retry logic and error handling.
Args:
endpoint: API endpoint path
payload: Request payload
stream: Whether to expect a streaming response
Returns:
Response data or None if request failed
Raises:
OllamaAPIException: If request fails after all retries
"""
url = f"{self.base_url}{endpoint}"
last_error = None
for attempt in range(self.config.max_retries):
try:
logger.debug(f"Making request to {url} (attempt {attempt + 1}/{self.config.max_retries})")
response = requests.post(
url,
json=payload,
timeout=self.config.request_timeout,
stream=stream
)
if response.status_code == 200:
if stream:
return response
return response.json()
elif response.status_code == 404:
error_msg = f"Model not found. Please ensure '{payload.get('model')}' is pulled."
logger.error(error_msg)
raise OllamaAPIException(error_msg)
elif response.status_code == 500:
last_error = f"Server error (500) on attempt {attempt + 1}"
logger.warning(last_error)
else:
error_msg = f"API returned status {response.status_code}: {response.text}"
logger.error(error_msg)
raise OllamaAPIException(error_msg)
except requests.exceptions.ConnectionError as e:
last_error = f"Connection error: {str(e)}"
logger.warning(f"Attempt {attempt + 1} failed: {last_error}")
except requests.exceptions.Timeout as e:
last_error = f"Request timeout: {str(e)}"
logger.warning(f"Attempt {attempt + 1} failed: {last_error}")
except requests.exceptions.RequestException as e:
last_error = f"Request exception: {str(e)}"
logger.warning(f"Attempt {attempt + 1} failed: {last_error}")
if attempt < self.config.max_retries - 1:
wait_time = self.config.retry_delay * (2 ** attempt)
logger.info(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
error_msg = f"Request failed after {self.config.max_retries} attempts. Last error: {last_error}"
logger.error(error_msg)
raise OllamaAPIException(error_msg)
def check_health(self) -> bool:
"""
Check if the Ollama service is running and accessible.
Returns:
True if service is healthy, False otherwise
"""
try:
response = requests.get(f"{self.base_url}/api/tags", timeout=5)
is_healthy = response.status_code == 200
logger.info(f"Health check: {'passed' if is_healthy else 'failed'}")
return is_healthy
except Exception as e:
logger.error(f"Health check failed: {e}")
return False
def generate(self, prompt: str, model: Optional[str] = None, **options) -> str:
"""
Generate text using the Ollama API.
Args:
prompt: Input prompt
model: Model name (uses default if not specified)
**options: Additional generation options
Returns:
Generated text
"""
model = model or self.config.default_model
payload = {
"model": model,
"prompt": prompt,
"stream": False,
"options": options or {
"temperature": self.config.temperature,
"num_predict": self.config.max_tokens
}
}
logger.info(f"Generating text with model {model}")
result = self._make_request("/api/generate", payload)
if result:
logger.info(f"Generated {result.get('eval_count', 0)} tokens in {result.get('eval_duration', 0) / 1e9:.2f}s")
return result.get('response', '')
return ""
def get_embedding(self, text: str, model: Optional[str] = None) -> np.ndarray:
"""
Get embedding vector for text.
Args:
text: Input text
model: Model name (uses embedding model if not specified)
Returns:
Numpy array containing the embedding
"""
model = model or self.config.embedding_model
payload = {
"model": model,
"prompt": text
}
logger.debug(f"Generating embedding for text of length {len(text)}")
result = self._make_request("/api/embeddings", payload)
if result and 'embedding' in result:
return np.array(result['embedding'])
raise OllamaAPIException("Failed to generate embedding")
class Document:
"""Represents a document in the knowledge base."""
def __init__(self, text: str, metadata: Optional[Dict] = None, doc_id: Optional[int] = None):
"""
Initialize a document.
Args:
text: Document text content
metadata: Optional metadata dictionary
doc_id: Optional document ID
"""
self.text = text
self.metadata = metadata or {}
self.id = doc_id
self.embedding = None
def __repr__(self):
return f"Document(id={self.id}, length={len(self.text)}, metadata={self.metadata})"
class DocumentStore:
"""Stores and manages documents with their embeddings."""
def __init__(self, client: OllamaClient):
"""
Initialize the document store.
Args:
client: OllamaClient instance
"""
self.client = client
self.documents: List[Document] = []
logger.info("Initialized document store")
def add_document(self, text: str, metadata: Optional[Dict] = None) -> int:
"""
Add a document to the store.
Args:
text: Document text
metadata: Optional metadata
Returns:
Document ID
"""
doc_id = len(self.documents)
doc = Document(text, metadata, doc_id)
try:
doc.embedding = self.client.get_embedding(text)
self.documents.append(doc)
logger.info(f"Added document {doc_id} with {len(text)} characters")
return doc_id
except Exception as e:
logger.error(f"Failed to add document: {e}")
raise
def add_documents_from_file(self, filepath: str, chunk_size: int = 1000) -> int:
"""
Add documents from a text file, splitting into chunks.
Args:
filepath: Path to the text file
chunk_size: Maximum characters per chunk
Returns:
Number of documents added
"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
chunks = []
current_chunk = ""
for paragraph in content.split('\n\n'):
if len(current_chunk) + len(paragraph) > chunk_size and current_chunk:
chunks.append(current_chunk.strip())
current_chunk = paragraph
else:
current_chunk += "\n\n" + paragraph if current_chunk else paragraph
if current_chunk:
chunks.append(current_chunk.strip())
count = 0
for i, chunk in enumerate(chunks):
if chunk:
self.add_document(
chunk,
metadata={"source": filepath, "chunk": i}
)
count += 1
logger.info(f"Added {count} document chunks from {filepath}")
return count
except Exception as e:
logger.error(f"Failed to add documents from file: {e}")
raise
def search(self, query: str, top_k: int = 3) -> List[Tuple[Document, float]]:
"""
Search for documents similar to the query.
Args:
query: Search query
top_k: Number of results to return
Returns:
List of (document, similarity_score) tuples
"""
if not self.documents:
logger.warning("No documents in store")
return []
try:
query_embedding = self.client.get_embedding(query)
similarities = []
for doc in self.documents:
if doc.embedding is not None:
similarity = np.dot(query_embedding, doc.embedding) / (
np.linalg.norm(query_embedding) * np.linalg.norm(doc.embedding)
)
similarities.append((doc, float(similarity)))
similarities.sort(key=lambda x: x[1], reverse=True)
results = similarities[:top_k]
logger.info(f"Search returned {len(results)} results for query: {query[:50]}...")
return results
except Exception as e:
logger.error(f"Search failed: {e}")
return []
def get_statistics(self) -> Dict:
"""
Get statistics about the document store.
Returns:
Dictionary with statistics
"""
total_chars = sum(len(doc.text) for doc in self.documents)
avg_chars = total_chars / len(self.documents) if self.documents else 0
return {
"total_documents": len(self.documents),
"total_characters": total_chars,
"average_characters": avg_chars,
"documents_with_embeddings": sum(1 for doc in self.documents if doc.embedding is not None)
}
class QuestionAnsweringSystem:
"""Complete question-answering system using RAG."""
def __init__(self, config: Config):
"""
Initialize the QA system.
Args:
config: Application configuration
"""
self.config = config
self.client = OllamaClient(config)
self.document_store = DocumentStore(self.client)
logger.info("Initialized Question-Answering System")
def add_knowledge(self, text: str, metadata: Optional[Dict] = None) -> int:
"""
Add knowledge to the system.
Args:
text: Knowledge text
metadata: Optional metadata
Returns:
Document ID
"""
return self.document_store.add_document(text, metadata)
def add_knowledge_from_file(self, filepath: str, chunk_size: int = 1000) -> int:
"""
Add knowledge from a file.
Args:
filepath: Path to file
chunk_size: Chunk size for splitting
Returns:
Number of documents added
"""
return self.document_store.add_documents_from_file(filepath, chunk_size)
def answer_question(self, question: str, verbose: bool = False) -> Dict:
"""
Answer a question using RAG.
Args:
question: The question to answer
verbose: Whether to include detailed information
Returns:
Dictionary with answer and metadata
"""
logger.info(f"Answering question: {question}")
start_time = time.time()
try:
relevant_docs = self.document_store.search(
question,
top_k=self.config.num_context_docs
)
if not relevant_docs:
logger.warning("No relevant documents found")
return {
"question": question,
"answer": "I don't have enough information to answer this question. Please add relevant documents to the knowledge base.",
"success": False
}
context_parts = []
for i, (doc, score) in enumerate(relevant_docs, 1):
context_parts.append(f"Context {i} (relevance: {score:.3f}):")
context_parts.append(doc.text)
context_parts.append("")
context = "\n".join(context_parts)
prompt = f"""Based on the following context, please answer the question accurately and concisely. If the context does not contain enough information to answer the question completely, acknowledge this and provide what information is available.
{context}
Question: {question}
Answer:"""
answer = self.client.generate(
prompt,
temperature=self.config.temperature,
num_predict=self.config.max_tokens
)
elapsed_time = time.time() - start_time
result = {
"question": question,
"answer": answer.strip(),
"success": True,
"elapsed_time": elapsed_time
}
if verbose:
result["context_documents"] = [
{
"text": doc.text[:200] + "..." if len(doc.text) > 200 else doc.text,
"similarity": score,
"metadata": doc.metadata
}
for doc, score in relevant_docs
]
logger.info(f"Question answered in {elapsed_time:.2f}s")
return result
except Exception as e:
logger.error(f"Failed to answer question: {e}")
return {
"question": question,
"answer": f"An error occurred while processing your question: {str(e)}",
"success": False
}
def get_system_status(self) -> Dict:
"""
Get system status and statistics.
Returns:
Dictionary with status information
"""
return {
"ollama_healthy": self.client.check_health(),
"document_store": self.document_store.get_statistics(),
"config": {
"model": self.config.default_model,
"embedding_model": self.config.embedding_model,
"temperature": self.config.temperature
}
}
def interactive_mode(qa_system: QuestionAnsweringSystem):
"""
Run the system in interactive mode.
Args:
qa_system: QuestionAnsweringSystem instance
"""
print("\n" + "=" * 80)
print("OLLAMA QUESTION-ANSWERING SYSTEM - Interactive Mode")
print("=" * 80)
print("\nCommands:")
print(" ask <question> - Ask a question")
print(" add <filepath> - Add documents from a file")
print(" status - Show system status")
print(" help - Show this help message")
print(" quit - Exit the program")
print("\n" + "=" * 80 + "\n")
while True:
try:
user_input = input("\nEnter command: ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', 'q']:
print("\nGoodbye!")
break
if user_input.lower() == 'help':
print("\nCommands:")
print(" ask <question> - Ask a question")
print(" add <filepath> - Add documents from a file")
print(" status - Show system status")
print(" help - Show this help message")
print(" quit - Exit the program")
continue
if user_input.lower() == 'status':
status = qa_system.get_system_status()
print("\nSystem Status:")
print(f" Ollama Service: {'Healthy' if status['ollama_healthy'] else 'Unhealthy'}")
print(f" Documents: {status['document_store']['total_documents']}")
print(f" Total Characters: {status['document_store']['total_characters']}")
print(f" Model: {status['config']['model']}")
continue
if user_input.lower().startswith('add '):
filepath = user_input[4:].strip()
if Path(filepath).exists():
print(f"\nAdding documents from {filepath}...")
count = qa_system.add_knowledge_from_file(filepath)
print(f"Successfully added {count} document chunks")
else:
print(f"\nError: File not found: {filepath}")
continue
if user_input.lower().startswith('ask '):
question = user_input[4:].strip()
if question:
print("\nProcessing your question...")
result = qa_system.answer_question(question, verbose=True)
print("\n" + "-" * 80)
print(f"Question: {result['question']}")
print("-" * 80)
print(f"Answer: {result['answer']}")
print("-" * 80)
if result['success']:
print(f"Response time: {result['elapsed_time']:.2f} seconds")
if 'context_documents' in result:
print(f"Context documents used: {len(result['context_documents'])}")
else:
print("\nPlease provide a question after 'ask'")
continue
print(f"\nUnknown command: {user_input}")
print("Type 'help' for available commands")
except KeyboardInterrupt:
print("\n\nGoodbye!")
break
except Exception as e:
logger.error(f"Error in interactive mode: {e}")
print(f"\nAn error occurred: {e}")
def main():
"""Main entry point for the application."""
parser = argparse.ArgumentParser(
description="Ollama-powered Document Question-Answering System"
)
parser.add_argument(
'--model',
default='llama2',
help='Model to use for generation (default: llama2)'
)
parser.add_argument(
'--url',
default='http://localhost:11434',
help='Ollama API base URL (default: http://localhost:11434)'
)
parser.add_argument(
'--temperature',
type=float,
default=0.3,
help='Generation temperature (default: 0.3)'
)
parser.add_argument(
'--add-file',
help='Add documents from file on startup'
)
parser.add_argument(
'--question',
help='Ask a single question and exit'
)
args = parser.parse_args()
config = Config(
ollama_base_url=args.url,
default_model=args.model,
embedding_model=args.model,
temperature=args.temperature
)
print("\nInitializing Question-Answering System...")
qa_system = QuestionAnsweringSystem(config)
if not qa_system.client.check_health():
print("\nError: Cannot connect to Ollama service")
print(f"Please ensure Ollama is running at {config.ollama_base_url}")
sys.exit(1)
print("System initialized successfully!")
if args.add_file:
print(f"\nAdding documents from {args.add_file}...")
try:
count = qa_system.add_knowledge_from_file(args.add_file)
print(f"Successfully added {count} document chunks")
except Exception as e:
print(f"Error adding documents: {e}")
sys.exit(1)
if args.question:
print(f"\nQuestion: {args.question}")
result = qa_system.answer_question(args.question)
print(f"\nAnswer: {result['answer']}")
sys.exit(0)
interactive_mode(qa_system)
if __name__ == "__main__":
main()
This complete application demonstrates professional-grade code organization and practices. The application is structured with clear separation of concerns, where each class has a specific responsibility. The Config dataclass centralizes all configuration parameters, making it easy to modify behavior without changing code. Comprehensive logging throughout the application aids in debugging and monitoring in production environments.
The error handling is robust and informative. Each component handles its own errors appropriately and provides meaningful error messages. The retry logic with exponential backoff ensures transient failures do not cause the application to fail unnecessarily. The health check functionality allows the application to verify the Ollama service is available before attempting to use it.
The Document and DocumentStore classes provide a clean abstraction for managing knowledge. Documents can be added individually or loaded from files, with automatic chunking for large documents. The search functionality uses cosine similarity to find relevant context, and the statistics method provides insights into the knowledge base.
The QuestionAnsweringSystem orchestrates the entire RAG pipeline, from retrieving context to generating answers. It provides both programmatic and command-line interfaces, making it suitable for integration into larger systems or standalone use. The interactive mode offers a user-friendly way to interact with the system, with commands for adding documents, asking questions, and checking system status.
The command-line argument parsing allows users to customize behavior without modifying code. Users can specify which model to use, adjust the temperature, add documents on startup, or ask a single question in batch mode. This flexibility makes the application suitable for various deployment scenarios.
This application represents production-ready code that could be deployed in real-world scenarios. It includes proper error handling, logging, configuration management, and a clean architecture that makes it easy to extend and maintain. You could enhance it further by adding features like persistent storage for the document store, support for multiple file formats, or a web interface, but the foundation is solid and follows best practices for Python application development.
