RAG Replacement Virtual Filesystem AI Assistant: A Step-by-Step Build Guide

1pip install chromadb openai pydantic fastapi uvicorn tiktoken
2pip install chromafs  # The virtual filesystem layer over ChromaDB

1git clone https://github.com/nichochar/chromafs.git
2cd chromafs
3pip install -e .

1docker run -d --name chromadb \
2  -p 8000:8000 \
3  -v chroma_data:/chroma/chroma \
4  chromadb/chroma:0.5.23

1mkdir vfs-assistant && cd vfs-assistant
2mkdir -p docs/en docs/zh-hant docs/zh-hans
3touch main.py filesystem.py ingest.py

1import chromadb
2from chromafs import ChromaFs
3
4def get_filesystem() -> ChromaFs:
5    """Initialize ChromaFs with a persistent ChromaDB backend."""
6    client = chromadb.HttpClient(host="localhost", port=8000)
7    fs = ChromaFs(
8        client=client,
9        collection_name="docs_vfs",
10        embedding_function=chromadb.utils.embedding_functions
11            .OpenAIEmbeddingFunction(
12                api_key="YOUR_OPENAI_KEY",
13                model_name="text-embedding-3-small"
14            )
15    )
16    return fs

1# Quick test
2fs = get_filesystem()
3print(fs.ls("/"))  
4# Output: []

1import os
2from pathlib import Path
3from filesystem import get_filesystem
4
5def ingest_docs(source_dir: str, vfs_root: str = "/"):
6    """Ingest a directory of markdown/text docs into the virtual filesystem."""
7    fs = get_filesystem()
8    source_path = Path(source_dir)
9    
10    files_ingested = 0
11    
12    for file_path in source_path.rglob("*.md"):
13        # Preserve relative path structure
14        relative = file_path.relative_to(source_path)
15        vfs_path = f"{vfs_root}{str(relative)}"
16        
17        # Create parent directories in VFS
18        parent_dir = str(Path(vfs_path).parent)
19        fs.mkdir(parent_dir, exist_ok=True)
20        
21        # Read and write file content
22        content = file_path.read_text(encoding="utf-8")
23        
24        # For large files, split by heading sections (not arbitrary chunks)
25        sections = split_by_headings(content)
26        
27        if len(sections) <= 1:
28            fs.write(vfs_path, content)
29            files_ingested += 1
30        else:
31            # Create a subdirectory for multi-section documents
32            doc_dir = vfs_path.replace(".md", "/")
33            fs.mkdir(doc_dir, exist_ok=True)
34            for i, (heading, body) in enumerate(sections):
35                section_path = f"{doc_dir}{i:02d}_{slugify(heading)}.md"
36                fs.write(section_path, f"# {heading}\n\n{body}")
37                files_ingested += 1
38    
39    return files_ingested
40
41def split_by_headings(content: str) -> list[tuple[str, str]]:
42    """Split markdown content by H2 headings, preserving structure."""
43    import re
44    parts = re.split(r"^## (.+)$", content, flags=re.MULTILINE)
45    
46    if len(parts) <= 1:
47        return [("Overview", content)]
48    
49    sections = []
50    # First part is content before any H2
51    if parts[0].strip():
52        sections.append(("Overview", parts[0].strip()))
53    
54    # Pair headings with their content
55    for i in range(1, len(parts), 2):
56        heading = parts[i]
57        body = parts[i + 1].strip() if i + 1 < len(parts) else ""
58        sections.append((heading, body))
59    
60    return sections
61
62def slugify(text: str) -> str:
63    import re
64    return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
65
66if __name__ == "__main__":
67    count = ingest_docs("./docs/en", vfs_root="/en/")
68    print(f"Ingested {count} files into virtual filesystem")

1python ingest.py
2# Output: Ingested 47 files into virtual filesystem

1import json
2from openai import OpenAI
3from filesystem import get_filesystem
4
5client = OpenAI(api_key="YOUR_OPENAI_KEY")
6fs = get_filesystem()
7
8TOOLS = [
9    {
10        "type": "function",
11        "function": {
12            "name": "ls",
13            "description": "List files and directories at a given path in the documentation filesystem.",
14            "parameters": {
15                "type": "object",
16                "properties": {
17                    "path": {
18                        "type": "string",
19                        "description": "The directory path to list, e.g. /en/api/"
20                    }
21                },
22                "required": ["path"]
23            }
24        }
25    },
26    {
27        "type": "function",
28        "function": {
29            "name": "read",
30            "description": "Read the full content of a file in the documentation filesystem.",
31            "parameters": {
32                "type": "object",
33                "properties": {
34                    "path": {
35                        "type": "string",
36                        "description": "The file path to read, e.g. /en/api/authentication.md"
37                    }
38                },
39                "required": ["path"]
40            }
41        }
42    },
43    {
44        "type": "function",
45        "function": {
46            "name": "search",
47            "description": "Semantic search across all documentation files. Returns top matches with paths.",
48            "parameters": {
49                "type": "object",
50                "properties": {
51                    "query": {
52                        "type": "string",
53                        "description": "Natural language search query"
54                    },
55                    "n_results": {
56                        "type": "integer",
57                        "description": "Number of results to return (default 5)"
58                    }
59                },
60                "required": ["query"]
61            }
62        }
63    }
64]
65
66SYSTEM_PROMPT = """You are a documentation assistant with access to a virtual filesystem 
67containing all product and regulatory documentation. 
68
69Before answering any question:
701. Use `search` to find relevant file paths
712. Use `ls` to explore the directory structure around those paths
723. Use `read` to get the full content of relevant files
734. Synthesize your answer from the actual file contents
74
75Always cite which files you referenced. If information spans multiple files, 
76read all of them before responding. Support Traditional Chinese, Simplified Chinese, 
77and English queries."""
78
79def execute_tool(name: str, args: dict) -> str:
80    """Execute a filesystem tool and return the result."""
81    try:
82        if name == "ls":
83            result = fs.ls(args["path"])
84            return json.dumps(result)
85        elif name == "read":
86            content = fs.read(args["path"])
87            return content
88        elif name == "search":
89            results = fs.search(
90                args["query"], 
91                n_results=args.get("n_results", 5)
92            )
93            return json.dumps([
94                {"path": r["path"], "preview": r["content"][:200]}
95                for r in results
96            ])
97    except Exception as e:
98        return f"Error: {str(e)}"
99
100def chat(user_message: str, history: list = None) -> str:
101    """Run a multi-turn conversation with filesystem tool use."""
102    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
103    if history:
104        messages.extend(history)
105    messages.append({"role": "user", "content": user_message})
106    
107    # Allow up to 10 tool-use rounds
108    for _ in range(10):
109        response = client.chat.completions.create(
110            model="gpt-4o",
111            messages=messages,
112            tools=TOOLS,
113            tool_choice="auto"
114        )
115        
116        msg = response.choices[0].message
117        messages.append(msg)
118        
119        if not msg.tool_calls:
120            return msg.content
121        
122        # Execute each tool call
123        for tool_call in msg.tool_calls:
124            result = execute_tool(
125                tool_call.function.name,
126                json.loads(tool_call.function.arguments)
127            )
128            messages.append({
129                "role": "tool",
130                "tool_call_id": tool_call.id,
131                "content": result
132            })
133    
134    return "Reached maximum tool-use iterations."
135
136if __name__ == "__main__":
137    answer = chat("What are the API rate limits for the enterprise tier?")
138    print(answer)

1python main.py
2# Output: Based on /en/api/03_rate-limits.md, the enterprise tier allows...

1from openai import OpenAI
2
3def detect_language(query: str) -> str:
4    """Fast language detection using the LLM itself."""
5    # For production, use a dedicated library like langdetect
6    # This is a simple approach for prototyping
7    import re
8    
9    # Check for CJK characters
10    cjk_chars = len(re.findall(r'[\u4e00-\u9fff]', query))
11    if cjk_chars > len(query) * 0.3:
12        # Distinguish Traditional vs Simplified
13        # Traditional Chinese has more complex characters
14        traditional_indicators = re.findall(
15            r'[\u9577\u8acb\u8a2a\u8b93\u6a23\u689d\u7d93\u904e]', 
16            query
17        )
18        if traditional_indicators:
19            return "zh-hant"
20        return "zh-hans"
21    
22    return "en"
23
24# Modify the system prompt to include language routing
25SYSTEM_PROMPT_MULTILINGUAL = """You are a documentation assistant. 
26Documentation is organized by language:
27- /en/ — English documentation
28- /zh-hant/ — Traditional Chinese (繁體中文)
29- /zh-hans/ — Simplified Chinese (简体中文)
30
31When a user asks in a specific language, search the corresponding 
32directory first, then fall back to /en/ if the content isn't available 
33in their language.
34
35Always respond in the same language as the user's query."""

1# api.py
2from fastapi import FastAPI, HTTPException
3from pydantic import BaseModel
4from main import chat
5import time
6
7app = FastAPI(title="VFS Documentation Assistant")
8
9class QueryRequest(BaseModel):
10    message: str
11    session_id: str | None = None
12    
13class QueryResponse(BaseModel):
14    answer: str
15    latency_ms: float
16    session_id: str
17
18# Simple in-memory session store (use Redis in production)
19sessions: dict[str, list] = {}
20
21@app.post("/query", response_model=QueryResponse)
22async def query_assistant(request: QueryRequest):
23    session_id = request.session_id or str(time.time())
24    history = sessions.get(session_id, [])
25    
26    start = time.perf_counter()
27    answer = chat(request.message, history=history)
28    latency = (time.perf_counter() - start) * 1000
29    
30    # Update session history
31    history.append({"role": "user", "content": request.message})
32    history.append({"role": "assistant", "content": answer})
33    sessions[session_id] = history[-20:]  # Keep last 20 messages
34    
35    return QueryResponse(
36        answer=answer,
37        latency_ms=round(latency, 2),
38        session_id=session_id
39    )
40
41@app.get("/health")
42async def health():
43    return {"status": "ok"}

1uvicorn api:app --host 0.0.0.0 --port 8080 --reload

1curl -X POST http://localhost:8080/query \
2  -H "Content-Type: application/json" \
3  -d '{"message": "How do I configure SSO for the enterprise plan?"}'
4
5# Output:
6# {
7#   "answer": "Based on /en/admin/sso-configuration.md, SSO setup requires...",
8#   "latency_ms": 1243.56,
9#   "session_id": "1719484823.456"
10# }

1import tiktoken
2
3enc = tiktoken.encoding_for_model("gpt-4o")
4
5def count_tokens(text: str) -> int:
6    return len(enc.encode(text))
7
8# In your execute_tool function, track cumulative tokens
9total_tokens_read = 0
10MAX_BUDGET = 80000  # Leave room for system prompt and output
11
12def execute_tool_with_budget(name: str, args: dict) -> str:
13    global total_tokens_read
14    result = execute_tool(name, args)
15    
16    if name == "read":
17        tokens = count_tokens(result)
18        total_tokens_read += tokens
19        if total_tokens_read > MAX_BUDGET:
20            return "Token budget exceeded. Summarize what you know and respond."
21    
22    return result

1# .github/workflows/reingest.yml
2on:
3  push:
4    paths: ["docs/**"]
5jobs:
6  reingest:
7    runs-on: ubuntu-latest
8    steps:
9      - uses: actions/checkout@v4
10      - run: pip install -r requirements.txt
11      - run: python ingest.py