Ghost in the Machine：以 FastAPI 建構注重私隱的本地 RAG 微服務

Hacktoberfest Open-Source AI Challenge: Ghost in the Machine Submission 👻🌼

This is a submission for the Hacktoberfest Open-Source AI Challenge Week 4: Ghost in the Machine

🌟 The Motivation: Why Open-Source AI Matters

Cloud-based AI APIs have simplified adding intelligence to applications, but they come with significant trade-offs: network latency, recurring subscription costs, rate limits, and vendor lock-in. More importantly, sending sensitive codebase or enterprise documentation to third-party endpoints creates serious data privacy concerns.

For Week 4's "Ghost in the Machine" challenge, I wanted to build a zero-cloud-dependency microservice that runs vector indexing and retrieval completely locally.

By keeping vector embedding and context matching on your local machine or private server, you retain 100% control over your data while enjoying sub-millisecond retrieval speeds.

🚀 What I Built: GhostInTheMachine RAG

GhostInTheMachine RAG is an asynchronous Retrieval-Augmented Generation (RAG) microservice built with FastAPI and Python 3.13.

Instead of pulling heavy external LLM packages, it features an in-memory term-frequency vector embedding engine with cosine similarity indexing built from scratch. It allows developers to ingest internal knowledge documents, store them securely in local memory, and query relevant contexts with sub-millisecond latency.

🛠 Tech Stack & Architecture

• Backend Engine: FastAPI (Asynchronous REST framework)
• Runtime & Server: Python 3.13 + Uvicorn ASGI Server
• Data Validation: Pydantic v2
• Vector Mechanics: In-Memory Term Frequency Vector Embedding + Cosine Similarity Scoring
• Developer Workflow: Visual Studio Code & DevRelay MCP Gateway Integration

💡 How It Works

The microservice processes incoming knowledge requests through a lean, three-stage local pipeline:

1. Document Ingestion (POST /ingest): Text content is tokenized, frequency-normalized into term vectors, and stored in the local vector store alongside metadata.
2. Vector Similarity Search (POST /query): Inbound queries are converted into normalized term vectors on the fly. The engine computes cosine similarity against indexed documents to retrieve top-$k$ matches.
3. Context Synthesis: Retrieved contexts are aggregated and synthesized into an immediate, zero-latency response payload without making external API calls.

💻 Core Implementation Highlights

Here is how the core local vector similarity engine is structured in app/services/rag_engine.py:

python
import time
import math
from typing import List, Dict

class SimpleVectorStore:
    def __init__(self):
        self.documents: List[Dict] = []

    def _tokenize(self, text: str) -> List[str]:
        return [w.lower() for w in text.split() if w.isalnum()]

    def _embed(self, text: str) -> Dict[str, float]:
        tokens = self._tokenize(text)
        if not tokens:
            return {}
        freq = {}
        for t in tokens:
            freq[t] = freq.get(t, 0) + 1.0
        norm = math.sqrt(sum(v ** 2 for v in freq.values()))
        return {k: v / norm for k, v in freq.items()}

    def _cosine_similarity(self, vec1: Dict[str, float], vec2: Dict[str, float]) -> float:
        score = 0.0
        for k, v in vec1.items():
            if k in vec2:
                score += v * vec2[k]
        return score

    def search(self, query: str, top_k: int = 3) -> List[Dict]:
        q_vec = self._embed(query)
        scored = []
        for doc in self.documents:
            sim = self._cosine_similarity(q_vec, doc["vector"])
            scored.append((sim, doc))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [doc for sim, doc in scored[:top_k] if sim > 0]