บทนำ

การพัฒนาซอฟต์แวร์ร่วมสมัยต้องการการเข้าถึงเอกสารและโค้ดอย่างรวดเร็ว โครงการนี้อธิบายวิธีสร้างระบบ RAG (Retrieval-Augmented Generation) สำหรับค้นหาและตอบคำถามเกี่ยวกับ GitHub repositories โดยใช้ HolySheep AI เป็นแกนหลัก ช่วยให้ทีมพัฒนาสามารถสืบค้น codebase ได้อย่างมีประสิทธิภาพ

ทำไมต้องย้ายมาใช้ HolySheep

ปัญหาที่พบกับ API เดิม

ข้อดีของ HolySheep

การตั้งค่าโครงสร้างโปรเจกต์

# requirements.txt
openai==1.12.0
chromadb==0.4.22
github3.py==3.3.0
tree-sitter==0.20.4
numpy==1.26.4
tiktoken==0.5.2
pydantic==2.6.1
python-dotenv==1.0.1
# config.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # HolySheep API Configuration
    HOLYSHEEP_API_KEY: str = "YOUR_HOLYSHEEP_API_KEY"
    HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
    
    # Model Settings
    EMBEDDING_MODEL: str = "deepseek-embedding-v3"
    LLM_MODEL: str = "deepseek-chat-v3"
    EMBEDDING_DIM: int = 3072
    EMBEDDING_BATCH_SIZE: int = 100
    
    # GitHub Settings
    GITHUB_TOKEN: Optional[str] = None
    REPO_OWNER: str = "your-org"
    REPO_NAME: str = "your-repo"
    
    # ChromaDB Settings
    CHROMA_PERSIST_DIR: str = "./chroma_db"
    COLLECTION_NAME: str = "codebase_embeddings"
    
    # RAG Settings
    CHUNK_SIZE: int = 800
    CHUNK_OVERLAP: int = 100
    TOP_K_RESULTS: int = 5
    
    class Config:
        env_file = ".env"

settings = Settings()

การดึงข้อมูลจาก GitHub Repository

# github_fetcher.py
import os
import github3
from github3.github import GitHub
from pathlib import Path
from typing import List, Generator
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class GitHubFetcher:
    """ดึงไฟล์โค้ดจาก GitHub repository"""
    
    def __init__(self, owner: str, repo: str, token: str = None):
        self.owner = owner
        self.repo = repo
        self.github = GitHub(token=token) if token else GitHub()
        
    def get_repository(self):
        """เชื่อมต่อกับ repository"""
        try:
            return self.github.repository(self.owner, self.repo)
        except Exception as e:
            logger.error(f"เชื่อมต่อ repository ล้มเหลว: {e}")
            raise
            
    def fetch_file_content(self, repo, path: str, ref: str = "main") -> dict:
        """ดึงเนื้อหาไฟล์เดี่ยว"""
        try:
            file_content = repo.file_contents(path, ref=ref)
            return {
                "path": path,
                "content": file_content.decoded.decode('utf-8'),
                "sha": file_content.sha,
                "size": len(file_content.decoded)
            }
        except Exception as e:
            logger.warning(f"ไม่สามารถดึง {path}: {e}")
            return None
            
    def fetch_all_files(self, repo, path: str = "") -> Generator[dict, None, None]:
        """ดึงไฟล์ทั้งหมดแบบ recursive"""
        try:
            contents = repo.contents(path)
            for content in contents:
                if content.type == "file":
                    # กรองเฉพาะไฟล์โค้ด
                    if self._is_code_file(content.name):
                        file_data = self.fetch_file_content(repo, content.path)
                        if file_data:
                            yield file_data
                elif content.type == "dir":
                    # ข้าม node_modules และไดเรกทอรีที่ไม่ต้องการ
                    if not self._should_skip_dir(content.name):
                        yield from self.fetch_all_files(repo, content.path)
        except Exception as e:
            logger.error(f"Error traversing {path}: {e}")
            
    def _is_code_file(self, filename: str) -> bool:
        """ตรวจสอบว่าเป็นไฟล์โค้ดหรือไม่"""
        code_extensions = {
            '.py', '.js', '.ts', '.jsx', '.tsx', '.java', '.cpp', '.c',
            '.h', '.hpp', '.go', '.rs', '.rb', '.php', '.swift', '.kt',
            '.scala', '.cs', '.vue', '.svelte', '.md', '.json', '.yaml', '.yml'
        }
        return any(filename.endswith(ext) for ext in code_extensions)
    
    def _should_skip_dir(self, dirname: str) -> bool:
        """ข้ามไดเรกทอรีที่ไม่ต้องการ"""
        skip_dirs = {
            'node_modules', '.git', '__pycache__', 'venv', '.venv',
            'dist', 'build', '.next', '.nuxt', 'coverage', '.pytest_cache',
            '.idea', '.vscode', 'vendor', 'target', 'bin', 'obj'
        }
        return dirname in skip_dirs or dirname.startswith('.')


การใช้งาน

fetcher = GitHubFetcher(
    owner="your-org",
    repo="your-repo",
    token=os.getenv("GITHUB_TOKEN")
)
repo = fetcher.get_repository()
file_count = 0

for file_data in fetcher.fetch_all_files(repo):
    print(f"ดึงไฟล์: {file_data['path']} ({file_data['size']} bytes)")
    file_count += 1
    
logger.info(f"ดึงไฟล์ทั้งหมด {file_count} ไฟล์")

การสร้าง Document Chunker สำหรับ Code

# document_processor.py
import re
from typing import List, Tuple
from dataclasses import dataclass
import logging

logger = logging.getLogger(__name__)

@dataclass
class CodeChunk:
    """โครงสร้างข้อมูลสำหรับ code chunk"""
    content: str
    metadata: dict
    chunk_id: str
    
class CodeChunker:
    """ตัดโค้ดเป็นส่วนย่อยสำหรับ embedding"""
    
    def __init__(self, chunk_size: int = 800, overlap: int = 100):
        self.chunk_size = chunk_size
        self.overlap = overlap
        
    def chunk_code(self, file_data: dict) -> List[CodeChunk]:
        """แบ่งโค้ดตามฟังก์ชันและคลาส"""
        content = file_data["content"]
        path = file_data["path"]
        chunks = []
        
        # ตรวจจับภาษาโปรแกรม
        language = self._detect_language(path)
        
        # สำหรับ Python ใช้ tree-sitter หรือ regex pattern
        if language == "python":
            chunks = self._chunk_python(content, path)
        elif language in ["javascript", "typescript"]:
            chunks = self._chunk_js_ts(content, path)
        else:
            # Fallback ใช้ character-based chunking
            chunks = self._chunk_by_characters(content, path)
            
        return chunks
    
    def _chunk_python(self, content: str, path: str) -> List[CodeChunk]:
        """แบ่ง Python code ตามฟังก์ชันและคลาส"""
        chunks = []
        
        # Pattern สำหรับฟังก์ชันและคลาส
        patterns = [
            (r'^class\s+\w+.*?(?=\nclass\s|\n\ndef\s|\Z)', 'class'),
            (r'^def\s+\w+.*?(?=\n^def\s|\n^class\s|\Z)', 'function'),
            (r'^async\s+def\s+\w+.*?(?=\n^async\s+def\s|\n^def\s|\n^class\s|\Z)', 'async_function'),
        ]
        
        processed_content = content
        
        for pattern, chunk_type in patterns:
            matches = list(re.finditer(pattern, content, re.MULTILINE | re.DOTALL))
            
            for i, match in enumerate(matches):
                chunk_content = match.group(0).strip()
                
                # ถ้า chunk ใหญ่เกินไป แบ่งย่อย
                if len(chunk_content) > self.chunk_size * 2:
                    sub_chunks = self._split_large_chunk(chunk_content)
                    for j, sub_chunk in enumerate(sub_chunks):
                        chunks.append(CodeChunk(
                            content=sub_chunk,
                            metadata={
                                "path": path,
                                "type": chunk_type,
                                "index": i,
                                "sub_index": j,
                                "language": "python"
                            },
                            chunk_id=f"{path}:{chunk_type}:{i}:{j}"
                        ))
                else:
                    chunks.append(CodeChunk(
                        content=chunk_content,
                        metadata={
                            "path": path,
                            "type": chunk_type,
                            "index": i,
                            "language": "python"
                        },
                        chunk_id=f"{path}:{chunk_type}:{i}"
                    ))
        
        # ถ้าไม่มีฟังก์ชันหรือคลาส ใช้ character-based
        if not chunks:
            chunks = self._chunk_by_characters(content, path)
            
        return chunks
    
    def _chunk_js_ts(self, content: str, path: str) -> List[CodeChunk]:
        """แบ่ง JavaScript/TypeScript code"""
        chunks = []
        
        patterns = [
            (r'export\s+(?:default\s+)?(?:class|function|const)\s+\w+.*?(?=\nexport\s|\Z)', 'export'),
            (r'(?:class|function)\s+\w+.*?(?=\n(?:class|function)\s|\Z)', 'definition'),
        ]
        
        for pattern, chunk_type in patterns:
            matches = list(re.finditer(pattern, content, re.DOTALL))
            for i, match in enumerate(matches):
                chunks.append(CodeChunk(
                    content=match.group(0).strip(),
                    metadata={
                        "path": path,
                        "type": chunk_type,
                        "index": i,
                        "language": "typescript" if path.endswith('.ts') else "javascript"
                    },
                    chunk_id=f"{path}:{chunk_type}:{i}"
                ))
                
        if not chunks:
            chunks = self._chunk_by_characters(content, path)
            
        return chunks
    
    def _chunk_by_characters(self, content: str, path: str) -> List[CodeChunk]:
        """แบ่งตามจำนวนตัวอักษร"""
        chunks = []
        lines = content.split('\n')
        current_chunk = []
        current_size = 0
        chunk_index = 0
        
        for line in lines:
            line_size = len(line)
            
            if current_size + line_size > self.chunk_size and current_chunk:
                chunks.append(CodeChunk(
                    content='\n'.join(current_chunk),
                    metadata={
                        "path": path,
                        "type": "block",
                        "index": chunk_index
                    },
                    chunk_id=f"{path}:block:{chunk_index}"
                ))
                
                # เก็บ overlap
                overlap_start = max(0, len(current_chunk) - 5)
                current_chunk = current_chunk[overlap_start:]
                current_size = sum(len(l) for l in current_chunk)
                chunk_index += 1
                
            current_chunk.append(line)
            current_size += line_size
            
        # เพิ่ม chunk สุดท้าย
        if current_chunk:
            chunks.append(CodeChunk(
                content='\n'.join(current_chunk),
                metadata={
                    "path": path,
                    "type": "block",
                    "index": chunk_index
                },
                chunk_id=f"{path}:block:{chunk_index}"
            ))
            
        return chunks
    
    def _split_large_chunk(self, content: str) -> List[str]:
        """แบ่ง chunk ขนาดใหญ่"""
        lines = content.split('\n')
        sub_chunks = []
        current = []
        current_size = 0
        
        for line in lines:
            if current_size + len(line) > self.chunk_size and current:
                sub_chunks.append('\n'.join(current))
                current = [line]
                current_size = len(line)
            else:
                current.append(line)
                current_size += len(line)
                
        if current:
            sub_chunks.append('\n'.join(current))
            
        return sub_chunks
    
    def _detect_language(self, path: str) -> str:
        """ตรวจจับภาษาโปรแกรมจากนามสกุลไฟล์"""
        ext_map = {
            '.py': 'python', '.js': 'javascript', '.ts': 'typescript',
            '.jsx': 'javascript', '.tsx': 'typescript', '.java': 'java',
            '.cpp': 'cpp', '.c': 'c', '.go': 'go', '.rs': 'rust'
        }
        for ext, lang in ext_map.items():
            if path.endswith(ext):
                return lang
        return 'unknown'

การสร้าง Embedding และ Vector Store

# vector_store.py
from openai import OpenAI
import chromadb
from chromadb.config import Settings as ChromaSettings
from typing import List, Optional
import hashlib
import logging

logger = logging.getLogger(__name__)

class CodeVectorStore:
    """จัดการ vector embedding และ ChromaDB"""
    
    def __init__(self, api_key: str, base_url: str, persist_dir: str):
        # ใช้ HolySheep แทน OpenAI โดยตรง
        self.client = OpenAI(
            api_key=api_key,
            base_url=base_url  # https://api.holysheep.ai/v1
        )
        
        self.chroma_client = chromadb.PersistentClient(
            path=persist_dir,
            settings=ChromaSettings(
                anonymized_telemetry=False,
                allow_reset=True
            )
        )
        
    def create_embeddings(self, texts: List[str], model: str = "deepseek-embedding-v3") -> List[List[float]]:
        """สร้าง embedding vectors ผ่าน HolySheep API"""
        try:
            response = self.client.embeddings.create(
                model=model,
                input=texts
            )
            
            embeddings = [item.embedding for item in response.data]
            logger.info(f"สร้าง embeddings สำเร็จ {len(embeddings)} รายการ")
            return embeddings
            
        except Exception as e:
            logger.error(f"สร้าง embedding ล้มเหลว: {e}")
            raise
            
    def setup_collection(self, collection_name: str, dimension: int = 3072):
        """สร้างหรือดึง collection"""
        try:
            collection = self.chroma_client.get_or_create_collection(
                name=collection_name,
                metadata={"dimension": dimension}
            )
            logger.info(f"Collection '{collection_name}' พร้อมใช้งาน")
            return collection
        except Exception as e:
            logger.error(f"สร้าง collection ล้มเหลว: {e}")
            raise
            
    def add_chunks(self, collection, chunks: List, batch_size: int = 100):
        """เพิ่ม chunks เข้า vector store"""
        embeddings = []
        documents = []
        metadatas = []
        ids = []
        
        for i, chunk in enumerate(chunks):
            documents.append(chunk.content)
            metadatas.append(chunk.metadata)
            ids.append(chunk.chunk_id)
            
            # Process เป็น batch
            if len(documents) >= batch_size or i == len(chunks) - 1:
                # สร้าง embeddings
                batch_embeddings = self.create_embeddings(documents)
                embeddings.extend(batch_embeddings)
                
                # เพิ่มเข้า ChromaDB
                collection.add(
                    embeddings=batch_embeddings,
                    documents=documents,
                    metadatas=metadatas,
                    ids=ids
                )
                
                logger.info(f"เพิ่ม {len(documents)} chunks เข้า vector store")
                
                # Reset batch
                documents = []
                metadatas = []
                ids = []
                
    def similarity_search(
        self,
        collection,
        query: str,
        top_k: int = 5,
        model: str = "deepseek-embedding-v3"
    ) -> List[dict]:
        """ค้นหา similarity"""
        # สร้าง query embedding
        query_embedding = self.create_embeddings([query], model)[0]
        
        # ค้นหาใน ChromaDB
        results = collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        # Format ผลลัพธ์
        formatted_results = []
        if results['documents'] and results['documents'][0]:
            for i, doc in enumerate(results['documents'][0]):
                formatted_results.append({
                    "content": doc,
                    "metadata": results['metadatas'][0][i],
                    "distance": results['distances'][0][i],
                    "id": results['ids'][0][i]
                })
                
        return formatted_results
        
    def reset_collection(self, collection_name: str):
        """ล้าง collection"""
        try:
            self.chroma_client.delete_collection(collection_name)
            logger.info(f"ลบ collection '{collection_name}' สำเร็จ")
        except Exception as e:
            logger.warning(f"ไม่สามารถลบ collection: {e}")


การใช้งาน

vector_store = CodeVectorStore(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    persist_dir="./chroma_db"
)

collection = vector_store.setup_collection("codebase_rag", dimension=3072)

RAG Engine สำหรับ Query

# rag_engine.py
from openai import OpenAI
from typing import List, Optional
import json
import logging

logger = logging.getLogger(__name__)

class CodeRAGEngine:
    """RAG Engine สำหรับตอบคำถามเกี่ยวกับ codebase"""
    
    SYSTEM_PROMPT = """คุณเป็นผู้เชี่ยวชาญด้านการวิเคราะห์โค้ด จงตอบคำถามโดยอิงจาก context ที่ให้มา
หากไม่แน่ใจ ให้บอกว่าไม่ทราบ ห้ามสร้างข้อมูลเท็จ ตอบเป็นภาษาไทย"""

    def __init__(
        self,
        api_key: str,
        base_url: str,
        vector_store,
        collection,
        model: str = "deepseek-chat-v3"
    ):
        self.client = OpenAI(
            api_key=api_key,
            base_url=base_url  # https://api.holysheep.ai/v1
        )
        self.vector_store = vector_store
        self.collection = collection
        self.model = model
        
    def ask(
        self,
        question: str,
        top_k: int = 5,
        max_context_tokens: int = 4000
    ) -> dict:
        """ถามคำถามเกี่ยวกับ codebase"""
        
        # 1. ค้นหา context ที่เกี่ยวข้อง
        search_results = self.vector_store.similarity_search(
            collection=self.collection,
            query=question,
            top_k=top_k
        )
        
        if not search_results:
            return {
                "answer": "ไม่พบข้อมูลที่เกี่ยวข้องใน codebase",
                "sources": []
            }
            
        # 2. สร้าง context string
        context = self._build_context(search_results, max_context_tokens)
        
        # 3. เรียก LLM
        response = self._generate_response(question, context)
        
        # 4. แนบ sources
        sources = [
            {
                "path": r["metadata"]["path"],
                "type": r["metadata"].get("type", "unknown"),
                "relevance": 1 - r["distance"]  # Convert distance to similarity
            }
            for r in search_results
        ]
        
        return {
            "answer": response,
            "sources": sources,
            "context_chunks": len(search_results)
        }
        
    def _build_context(self, results: List[dict], max_tokens: int) -> str:
        """สร้าง context string จากผลการค้นหา"""
        context_parts = []
        current_tokens = 0
        
        for result in results:
            chunk = result["content"]
            path = result["metadata"]["path"]
            chunk_type = result["metadata"].get("type", "code")
            
            # ประมาณ token count (1 token ≈ 4 characters)
            chunk_tokens = len(chunk) // 4
            
            if current_tokens + chunk_tokens > max_tokens:
                break
                
            context_parts.append(
                f"ไฟล์: {path} ({chunk_type})\n``\n{chunk}\n``"
            )
            current_tokens += chunk_tokens
            
        return "\n\n---\n\n".join(context_parts)
        
    def _generate_response(self, question: str, context: str) -> str:
        """เรียก LLM เพื่อสร้างคำตอบ"""
        messages = [
            {"role": "system", "content": self.SYSTEM_PROMPT},
            {"role": "user", "content": f"Context:\n{context}\n\nคำถาม: {question}"}
        ]
        
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.3,
                max_tokens=1000
            )
            
            return response.choices[0].message.content
            
        except Exception as e:
            logger.error(f"LLM call failed: {e}")
            raise


การใช้งาน

rag_engine = CodeRAGEngine(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
    vector_store=vector_store,
    collection=collection,
    model="deepseek-chat-v3"
)


ถามคำถาม

result = rag_engine.ask("ฟังก์ชัน login อยู่ที่ไหน และทำงานอย่างไร?")
print(f"คำตอบ: {result['answer']}")
print(f"แหล่งอ้างอิง: {len(result['sources'])} รายการ")

Pipeline หลักสำหรับ Indexing และ Query

# main.py
import os
from dotenv import load_dotenv
from github_fetcher import GitHubFetcher
from document_processor import CodeChunker
from vector_store import CodeVectorStore
from rag_engine import CodeRAGEngine
from config import settings
import logging

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

class CodebaseRAGPipeline:
    """Pipeline หลักสำหรับ Indexing และ Query codebase"""
    
    def __init__(self):
        load_dotenv()
        
        # Initialize components
        self.fetcher = GitHubFetcher(
            owner=settings.REPO_OWNER,
            repo=settings.REPO_NAME,
            token=settings.GITHUB_TOKEN
        )
        
        self.chunker = CodeChunker(
            chunk_size=settings.CHUNK_SIZE,
            overlap=settings.CHUNK_OVERLAP
        )
        
        self.vector_store = CodeVectorStore(
            api_key=settings.HOLYSHEEP_API_KEY,
            base_url=settings.HOLYSHEEP_BASE_URL,
            persist_dir=settings.CHROMA_PERSIST_DIR
        )
        
        self.collection = self.vector_store.setup_collection(
            settings.COLLECTION_NAME,
            dimension=settings.EMBEDDING_DIM
        )
        
        self.rag_engine = CodeRAGEngine(
            api_key=settings.HOLYSHEEP_API_KEY,
            base_url=settings.HOLYSHEEP_BASE_URL,
            vector_store=self.vector_store,
            collection=self.collection,
            model=settings.LLM_MODEL
        )
        
    def index_repository(self, ref: str = "main", force_rebuild: bool = False):
        """Index ทั้ง repository"""
        
        if force_rebuild:
            logger.info("ล้าง collection เดิม...")
            self.vector_store.reset_collection(settings.COLLECTION_NAME)
            self.collection = self.vector_store.setup_collection(
                settings.COLLECTION_NAME,
                dimension=settings.EMBEDDING_DIM
            )
            
        # ดึงไฟล์จาก GitHub
        repo = self.fetcher.get_repository()
        logger.info(f"เริ่ม index repository: {settings.REPO_OWNER}/{settings.REPO_NAME}")
        
        all_chunks = []
        file_count = 0
        
        for file_data in self.fetcher.fetch_all_files(repo):
            logger.info(f"ประมวลผล: {file_data['path']}")
            
            # แบ่งเป็น chunks
            chunks = self.chunker.chunk_code(file_data)
            all_chunks.extend(chunks)
            file_count += 1
            
        logger.info(f"ได้ chunks ทั้งหมด {len(all_chunks)} ชิ้น จาก {file_count} ไฟล์")
        
        # เพิ่มเข้า vector store
        self.vector_store.add_chunks(
            self.collection,
            all_chunks,
            batch_size=settings.EMBEDD

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง