GPT-OSS Review: OpenAI's Free Model

import ollama
# Simple one-off generation
response = ollama.generate(
    model='gpt-oss:20b',
    prompt='What are three real-world use cases for an AI model that runs locally?'
)
print(response['response'])

import cv2
import ollama
from ultralytics import YOLO
import json
from datetime import datetime

class VisualQASystem:
    def __init__(self, yolo_model_path="yolov8n.pt", gpt_model="gpt-oss:20b"):
        # Initialize YOLO model
        self.yolo_model = YOLO(yolo_model_path)
        self.gpt_model = gpt_model
        # Class names for COCO dataset (YOLOv8 default)
        self.class_names = self.yolo_model.names
        
    def detect_objects(self, image):
        """Run YOLO detection on image"""
        results = self.yolo_model(image)
        detections = []
        for result in results:
            boxes = result.boxes
            if boxes is not None:
                for box in boxes:
                    # Extract detection data
                    class_id = int(box.cls[0])
                    confidence = float(box.conf[0])
                    coords = box.xyxy[0].tolist()  # [x1, y1, x2, y2]
                    detection = {
                        'class': self.class_names[class_id],
                        'confidence': round(confidence, 3),
                        'bbox': coords,
                        'center': [(coords[0] + coords[2])/2, (coords[1] + coords[3])/2]
                    }
                    detections.append(detection)
        return detections
    
    def format_detection_data(self, detections, image_context=""):
        """Convert YOLO detections to structured text for GPT-OSS"""
        if not detections:
            return "No objects detected in the current frame."
        detections = sorted(detections, key=lambda x: x['confidence'], reverse=True)
        detection_text = f"Image Analysis - {datetime.now().strftime('%H:%M:%S')}\n"
        if image_context:
            detection_text += f"Context: {image_context}\n"
        detection_text += f"Objects Detected ({len(detections)} total):\n"
        for i, det in enumerate(detections, 1):
            detection_text += f"{i}. {det['class']} (confidence: {det['confidence']:.1%})\n"
            detection_text += f"   Location: center at ({det['center'][0]:.0f}, {det['center'][1]:.0f})\n"
        return detection_text
    
    def generate_description(self, detection_data, query_type="describe"):
        """Generate natural language response using GPT-OSS 20B"""
        prompts = {
            "describe": f"""Analyze this object detection data and provide a natural, conversational description of what's happening in the scene:
{detection_data}
Provide a clear, human-friendly description focusing on the most important objects and their relationships.""",
            "alert": f"""You are a security monitoring system. Analyze the following object detection data and generate appropriate alerts or notifications:
{detection_data}
Focus on:
- Unusual or suspicious activities
- Safety concerns
- Objects that shouldn't be in certain areas
- Any anomalies that require attention
Provide concise, actionable alerts.""",
            "count": f"""Analyze the detection data and provide a summary count of different object types:
{detection_data}
Provide a structured count and brief analysis of the distribution of objects.""",
            "safety": f"""Evaluate this scene for potential safety hazards:
{detection_data}
Identify any safety concerns, potential risks, or recommendations for the observed scene."""
        }
        prompt = prompts.get(query_type, prompts["describe"])
        try:
            response = ollama.generate(
                model=self.gpt_model,
                prompt=prompt,
                options={
                    'temperature': 0.3,
                    'top_p': 0.9,
                    'num_predict': 2000
                }
            )
            return response['response'].strip()
        except Exception as e:
            return f"Error generating response: {str(e)}"
    
    def process_frame(self, image, query_type="describe", context=""):
        """Complete pipeline: detect -> format -> generate description"""
        detections = self.detect_objects(image)
        detection_text = self.format_detection_data(detections, context)
        description = self.generate_description(detection_text, query_type)
        return {
            'detections': detections,
            'detection_text': detection_text,
            'description': description,
            'timestamp': datetime.now()
        }

# Usage Examples
def main():
    # Initialize the system
    vqa_system = VisualQASystem()
    # Example 1: Process single image
    image_path = "ADE_val_00000022.jpg"
    image = cv2.imread(image_path)
    result = vqa_system.process_frame(
        image, 
        query_type="describe", 
        context="Street View"
    )
    print("=== DETECTION RESULTS ===")
    print(result['detection_text'])
    print("\n=== NATURAL LANGUAGE DESCRIPTION ===")
    print(result['description'])

if __name__ == "__main__":
    main()

import os
import json
import ollama
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, TextLoader, DirectoryLoader
from langchain_community.document_loaders import Docx2txtLoader
import uuid
from datetime import datetime
from typing import List, Dict, Any

class RAGExpertSystem:
    def __init__(self, 
                 knowledge_base_path="./knowledge_base",
                 vector_db_path="./vector_db",
                 gpt_model="gpt-oss:20b",
                 embedding_model="all-MiniLM-L6-v2"):
        # Initialize components
        self.knowledge_base_path = knowledge_base_path
        self.vector_db_path = vector_db_path
        self.gpt_model = gpt_model
        # Initialize embedding model
        print("Loading embedding model...")
        self.embedding_model = SentenceTransformer(embedding_model)
        # Initialize ChromaDB client
        print("Initializing vector database...")
        self.chroma_client = chromadb.PersistentClient(path=vector_db_path)
        self.collection_name = "expert_knowledge"
        # Create or get collection
        try:
            self.collection = self.chroma_client.get_collection(self.collection_name)
            print(f"Loaded existing collection with {self.collection.count()} documents")
        except:
            self.collection = self.chroma_client.create_collection(
                name=self.collection_name,
                metadata={ "description": "Expert knowledge base for RAG system" }
            )
            print("Created new collection")
        # Initialize text splitter
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            length_function=len,
        )
        print("RAG Expert System initialized successfully!")
    def load_documents(self, file_paths: List[str] = None):
        """Load documents from specified paths or directory"""
        documents = []
        if file_paths:
            # Load specific files
            for file_path in file_paths:
                print(f"Loading: {file_path}")
                if file_path.endswith('.pdf'):
                    loader = PyPDFLoader(file_path)
                elif file_path.endswith('.docx'):
                    loader = Docx2txtLoader(file_path)
                elif file_path.endswith('.txt'):
                    loader = TextLoader(file_path)
                else:
                    print(f"Unsupported file type: {file_path}")
                    continue
                docs = loader.load()
                documents.extend(docs)
        else:
            # Load all documents from knowledge base directory
            if os.path.exists(self.knowledge_base_path):
                print(f"Loading documents from {self.knowledge_base_path}")
                # Load PDFs
                pdf_loader = DirectoryLoader(
                    self.knowledge_base_path,
                    glob="**/*.pdf",
                    loader_cls=PyPDFLoader
                )
                # Load text files
                txt_loader = DirectoryLoader(
                    self.knowledge_base_path,
                    glob="**/*.txt",
                    loader_cls=TextLoader
                )
                documents.extend(pdf_loader.load())
                documents.extend(txt_loader.load())
            else:
                print(f"Knowledge base directory {self.knowledge_base_path} not found")
        print(f"Loaded {len(documents)} documents")
        return documents
    def process_and_store_documents(self, documents):
        """Split documents into chunks and store in vector database"""
        print("Processing documents...")
        all_chunks = []
        all_embeddings = []
        all_metadatas = []
        all_ids = []
        for doc_idx, document in enumerate(documents):
            # Split document into chunks
            chunks = self.text_splitter.split_text(document.page_content)
            for chunk_idx, chunk in enumerate(chunks):
                # Create embedding
                embedding = self.embedding_model.encode(chunk).tolist()
                # Create metadata
                metadata = {
                    "source": document.metadata.get("source", f"document_{doc_idx}"),
                    "chunk_id": chunk_idx,
                    "timestamp": datetime.now().isoformat(),
                    "length": len(chunk)
                }
                # Create unique ID
                doc_id = str(uuid.uuid4())
                all_chunks.append(chunk)
                all_embeddings.append(embedding)
                all_metadatas.append(metadata)
                all_ids.append(doc_id)
        # Store in ChromaDB
        print(f"Storing {len(all_chunks)} chunks in vector database...")
        self.collection.add(
            documents=all_chunks,
            embeddings=all_embeddings,
            metadatas=all_metadatas,
            ids=all_ids
        )
        print(f"Successfully stored {len(all_chunks)} chunks")
        return len(all_chunks)
    def retrieve_relevant_context(self, query: str, n_results: int = 5) -> List[Dict]:
        """Retrieve relevant document chunks for a query"""
        # Create query embedding
        query_embedding = self.embedding_model.encode(query).tolist()
        # Search for similar documents
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=n_results,
            include=["documents", "metadatas", "distances"]
        )
        # Format results
        context_chunks = []
        for i in range(len(results['documents'][0])):
            context_chunks.append({
                'content': results['documents'][0][i],
                'metadata': results['metadatas'][0][i],
                'similarity': 1 - results['distances'][0][i]
            })
        return context_chunks
    def generate_expert_response(self, query: str, context_chunks: List[Dict]) -> str:
        """Generate response using GPT-OSS with retrieved context"""
        # Format context
        context_text = "\n\n".join([
            f"[Source: {chunk['metadata']['source']}]\n{chunk['content']}"
            for chunk in context_chunks
        ])
        # Create expert prompt
        prompt = f"""You are an expert assistant with access to a specialized knowledge base. Use the provided context to give accurate, detailed, and helpful responses.
CONTEXT FROM KNOWLEDGE BASE:
{context_text}
USER QUESTION: {query}
INSTRUCTIONS:
- Answer based primarily on the provided context
- If the context doesn't contain enough information, clearly state what's missing
- Provide specific references to sources when possible
- Give detailed, expert-level explanations
- If you find conflicting information, acknowledge it
EXPERT RESPONSE:"""
        try:
            response = ollama.generate(
                model=self.gpt_model,
                prompt=prompt,
                options={
                    'temperature': 0.1,
                    'top_p': 0.9,
                    'num_predict': 1000
                }
            )
            return response['response'].strip()
        except Exception as e:
            return f"Error generating response: {str(e)}"
    def chat(self, query: str, n_results: int = 5) -> Dict[str, Any]:
        """Complete RAG pipeline: retrieve + generate"""
        print(f"Processing query: {query}")
        # Step 1: Retrieve relevant context
        context_chunks = self.retrieve_relevant_context(query, n_results)
        # Step 2: Generate expert response
        response = self.generate_expert_response(query, context_chunks)
        return {
            'query': query,
            'response': response,
            'context_used': context_chunks,
            'timestamp': datetime.now().isoformat(),
            'sources': list(set([chunk['metadata']['source'] for chunk in context_chunks]))
        }
    def add_document_from_text(self, text: str, source_name: str):
        """Add a single document from text"""
        # Create a document-like object
        class SimpleDoc:
            def __init__(self, content, source):
                self.page_content = content
                self.metadata = { "source": source }
        
        doc = SimpleDoc(text, source_name)
        self.process_and_store_documents([doc])
        print(f"Added document: {source_name}")
    def get_knowledge_base_stats(self):
        """Get statistics about the knowledge base"""
        count = self.collection.count()
        # Get sample of metadata to analyze sources
        if count > 0:
            sample = self.collection.get(limit=min(100, count), include=["metadatas"])
            sources = set([meta['source'] for meta in sample['metadatas']])
            return {
                'total_chunks': count,
                'unique_sources': len(sources),
                'sample_sources': list(sources)[:10]
            }
        return {'total_chunks': 0, 'unique_sources': 0, 'sample_sources': []}

def main():
    # Initialize RAG system
    rag_system = RAGExpertSystem()
    # Option 1: Load documents from directory
    print("\n=== Loading Knowledge Base ===")
    documents = rag_system.load_documents()
    if documents:
        chunks_stored = rag_system.process_and_store_documents(documents)
        print(f"Knowledge base ready with {chunks_stored} chunks!")
    else:
        print("No documents found. You can add documents manually or place files in ./knowledge_base/")
    stats = rag_system.get_knowledge_base_stats()
    print(f"\n=== Knowledge Base Stats ===")
    print(f"Total chunks: {stats['total_chunks']}")
    print(f"Unique sources: {stats['unique_sources']}")
    if stats['sample_sources']:
        print(f"Sample sources: {', '.join(stats['sample_sources'])}")
    print("\n=== RAG Expert Chat ===")
    print("Ask questions about your knowledge base. Type 'quit' to exit.")
    while True:
        query = input("\n🤖 Your Question: ").strip()
        if query.lower() in ['quit', 'exit', 'q']:
            break
        if not query:
            continue
        result = rag_system.chat(query, n_results=3)
        print(f"\n📚 Expert Response:")
        print(result['response'])
        print(f"\n📄 Sources Used:")
        for source in result['sources']:
            print(f"  - {source}")
        print(f"\n🔍 Context Chunks (for debugging):")
        for i, chunk in enumerate(result['context_used'], 1):
            print(f"  {i}. Similarity: {chunk['similarity']:.3f} | Source: {chunk['metadata']['source']}")
            print(f"     Preview: {chunk['content'][:100]}...")

def setup_sample_knowledge_base():
    """Create a sample knowledge base for testing"""
    rag_system = RAGExpertSystem()
    # Add some sample expert knowledge
    sample_docs = [
        {
            "text": """
Machine Learning Model Deployment Best Practices:
1. Model Versioning: Always version your models using tools like MLflow or DVC
2. A/B Testing: Implement gradual rollouts with A/B testing framework
3. Monitoring: Set up model drift detection and performance monitoring
4. Containerization: Use Docker for consistent deployment environments
5. CI/CD: Automate testing and deployment pipelines
6. Rollback Strategy: Have a quick rollback mechanism for failed deployments
""",
            "source": "ml_deployment_guide.txt"
        },
        {
            "text": """
Computer Vision Pipeline Optimization:
Performance Optimization Techniques:
- Use optimized inference engines like ONNX Runtime or TensorRT
- Implement batch processing for multiple images
- Apply model quantization to reduce memory usage
- Use GPU acceleration with CUDA when available
- Implement caching for repeated inference requests
Quality Assurance:
- Validate input image formats and resolutions
- Implement confidence thresholds for predictions
- Use ensemble methods for critical applications
- Monitor prediction latency and accuracy metrics
""",
            "source": "cv_optimization_manual.txt"
        }
    ]
    for doc in sample_docs:
        rag_system.add_document_from_text(doc["text"], doc["source"])
    return rag_system

if __name__ == "__main__":
    # Option 1: Use with your own documents
    main()
    # Option 2: Use with sample knowledge base
    # rag_system = setup_sample_knowledge_base()
    # result = rag_system.chat("How do I optimize computer vision models for production?")
    # print(result['response'])