Build AI-Powered Apps

version: '3.8'

services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER: raguser
      POSTGRES_PASSWORD: ragpass123
      POSTGRES_DB: ragdb
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  ollama:
    image: ollama/ollama:latest
    ports:
      - "11434:11434"
    volumes:
      - ollama_data:/root/.ollama

  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma_data:/chroma/chroma
    environment:
      - IS_PERSISTENT=TRUE

  api:
    build: ./app
    ports:
      - "8080:8080"
    depends_on:
      - postgres
      - chromadb
      - ollama
    environment:
      - DATABASE_URL=postgresql://raguser:ragpass123@postgres:5432/ragdb
      - CHROMA_HOST=chromadb
      - OLLAMA_HOST=ollama

volumes:
  postgres_data:
  ollama_data:
  chroma_data:

from sqlalchemy import Column, Integer, String, Text, DateTime
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

Base = declarative_base()

# Database Table Definition
class CodeSnippet(Base):
    __tablename__ = "code_snippets"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), nullable=False)
    description = Column(Text)
    code = Column(Text, nullable=False)
    language = Column(String(50), default="python")
    tags = Column(String(255))
    created_at = Column(DateTime, default=datetime.utcnow)

# API Input Schema (what users send)
class SnippetCreate(BaseModel):
    title: str
    description: Optional[str] = None
    code: str
    language: str = "python"
    tags: Optional[str] = None

# API Output Schema (what API returns)
class SnippetResponse(BaseModel):
    id: int
    title: str
    description: Optional[str]
    code: str
    language: str
    tags: Optional[str]
    created_at: datetime
    
    class Config:
        from_attributes = True

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db, engine
from models import Base, SnippetCreate, SnippetResponse, CodeSnippet

Base.metadata.create_all(bind=engine)  # Create tables!

app = FastAPI(title="RAG Snippets API")

# CREATE - Add new snippet
@app.post("/snippets/", response_model=SnippetResponse)
def create_snippet(snippet: SnippetCreate, db: Session = Depends(get_db)):
    db_snippet = CodeSnippet(**snippet.model_dump())
    db.add(db_snippet)
    db.commit()
    db.refresh(db_snippet)
    return db_snippet

# READ - Get all snippets
@app.get("/snippets/", response_model=list[SnippetResponse])
def read_snippets(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return db.query(CodeSnippet).offset(skip).limit(limit).all()

# READ - Get one snippet
@app.get("/snippets/{snippet_id}", response_model=SnippetResponse)
def read_snippet(snippet_id: int, db: Session = Depends(get_db)):
    snippet = db.query(CodeSnippet).filter(CodeSnippet.id == snippet_id).first()
    if not snippet:
        raise HTTPException(status_code=404, detail="Not found")
    return snippet

# DELETE - Remove snippet
@app.delete("/snippets/{snippet_id}")
def delete_snippet(snippet_id: int, db: Session = Depends(get_db)):
    snippet = db.query(CodeSnippet).filter(CodeSnippet.id == snippet_id).first()
    if not snippet:
        raise HTTPException(status_code=404, detail="Not found")
    db.delete(snippet)
    db.commit()
    return {"message": "Deleted!"}

import chromadb
import os

class ChromaService:
    def __init__(self):
        # Connect to ChromaDB server
        self.client = chromadb.HttpClient(
            host=os.getenv("CHROMA_HOST", "localhost"),
            port=8000
        )
        # Get or create collection
        self.collection = self.client.get_or_create_collection(
            name="code_snippets"
        )
    
    def add_snippet(self, snippet_id: int, title: str, code: str, 
                      description: str = "", tags: str = ""):
        # Combine all text for embedding
        document = f"{title}\n{description}\n{code}"
        
        self.collection.add(
            documents=[document],           # Text to embed
            metadatas=[{"title": title, "tags": tags}],  # Extra info
            ids=[f"snippet_{snippet_id}"]   # Unique ID
        )
    
    def search_snippets(self, query: str, n_results: int = 5):
        # Find similar documents
        results = self.collection.query(
            query_texts=[query],    # Your question
            n_results=n_results     # Top N matches
        )
        return results
    
    def delete_snippet(self, snippet_id: int):
        self.collection.delete(ids=[f"snippet_{snippet_id}"])

from langchain_community.llms import Ollama
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import chromadb, os

class RAGService:
    def __init__(self):
        # 1. Initialize Ollama LLM
        self.llm = Ollama(
            model="llama2",
            base_url=f"http://{os.getenv('OLLAMA_HOST', 'localhost')}:11434"
        )
        
        # 2. Initialize embeddings
        self.embeddings = OllamaEmbeddings(
            model="llama2",
            base_url=f"http://{os.getenv('OLLAMA_HOST', 'localhost')}:11434"
        )
        
        # 3. Connect to ChromaDB
        self.vectorstore = Chroma(
            collection_name="code_snippets",
            embedding_function=self.embeddings,
            client=chromadb.HttpClient(
                host=os.getenv("CHROMA_HOST", "localhost"), port=8000
            )
        )
        
        # 4. Create retriever
        self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
        
        # 5. Custom prompt template
        self.prompt = PromptTemplate(
            input_variables=["context", "question"],
            template="""Use these code snippets to answer:

Context: {context}

Question: {question}

Answer: """
        )
        
        # 6. Build the RAG chain
        self.chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            chain_type_kwargs={"prompt": self.prompt},
            return_source_documents=True
        )
    
    def query(self, question: str):
        result = self.chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.metadata for doc in result["source_documents"]]
        }