向量文档搜索系统

HTTP API (FastAPI, port 8391)
    ↓
Query Engine: embed query → search all shards → merge by score → fetch text
    ↓
┌────────────────────────────────────────────────────────────┐
│  Per-Backend Shards                                        │
│                                                            │
│  data/shards/wikipedia-en/     data/shards/my-docs/        │
│    ├── faiss.index               ├── faiss.index           │
│    └── refs.sqlite               └── refs.sqlite           │
│                                                            │
│  Each shard has its own FAISS index + SQLite ref store.    │
│  Shards can be ingested, dropped, and rebuilt              │
│  independently.                                            │
└────────────────────────────────────────────────────────────┘
    ↓
Storage Backend Plugins
┌─────┬───────────┬──────┐
│ ZIM │ Plaintext │ ...  │
└─────┴───────────┴──────┘

cd engine

# Using pipenv (recommended)
pipenv install

# Or with pip
pip install -e .

# With ZIM support
pip install -e ".[zim]"

mkdir -p data
cp engine/backends.example.json data/backends.json
# Edit data/backends.json with your actual paths

{
  "backends": [
    {
      "id": "wikipedia-en",
      "type": "zim",
      "description": "English Wikipedia — full text of all articles",
      "paths": ["/media/archive/wikipedia_en.zim"],
      "min_text_length": 200
    },
    {
      "id": "notes",
      "type": "plaintext",
      "description": "Personal notes and documentation",
      "paths": ["/home/user/notes"]
    }
  ]
}

pipenv run python -m potatosearch.cli ingest --backend wikipedia-en

pipenv run python -m potatosearch.cli ingest --backend wikipedia-en --train

pipenv run python -m potatosearch.cli query "How does photosynthesis work?" --top-k 5

pipenv run potatosearch-server
# or: pipenv run python -m potatosearch.api.server

curl -X POST http://localhost:8391/api/query \
  -H "Content-Type: application/json" \
  -d '{"question": "How does photosynthesis work?", "top_k": 5}'

curl -X POST http://localhost:8391/api/query \
  -H "Content-Type: application/json" \
  -d '{"question": "How does photosynthesis work?", "top_k": 5, "backends": ["wikipedia-en"]}'

# Build the frontend
cd ui
npm install
npm run build

# Start the engine (UI is served automatically from ui/dist/)
cd ../engine
pipenv run potatosearch-server

# Terminal 1: start the engine
cd engine
pipenv run potatosearch-server

# Terminal 2: start Vite dev server (proxies API calls to engine)
cd ui
npm run dev

docker build -t potatosearch .
docker run -v ./data:/data -p 8391:8391 potatosearch

docker build --target engine -t potatosearch-engine .
docker run -v ./data:/data -p 8391:8391 potatosearch-engine

{
  "mcpServers": {
    "potatosearch": {
      "type": "streamable-http",
      "url": "http://your-server:8391/mcp"
    }
  }
}

# 1. Drop the old shard
pipenv run python -m potatosearch.cli drop --backend wikipedia-en

# 2. Update backends.json to point to the new ZIM file

# 3. Reingest
pipenv run python -m potatosearch.cli ingest --backend wikipedia-en --train

pipenv run python -m potatosearch.cli stats

  notes: 1200 chunks, 1200 vectors
  stackexchange: 5400000 chunks, 5400000 vectors
  wikipedia-en: 22000000 chunks, 22000000 vectors
Total: 27401200 chunks, 27401200 vectors

# engine/potatosearch/backends/my_backend.py
from pathlib import Path
from typing import Iterator
from potatosearch.core import Document, StorageBackend

class MyBackend(StorageBackend):

    def __init__(self, some_config: str, *, backend_id: str = "myformat"):
        self._backend_id = backend_id
        self._config = some_config

    @property
    def name(self) -> str:
        return self._backend_id

    def iterate_documents(self) -> Iterator[Document]:
        """
        Yield every document from your source. Called during ingestion.
        Each Document has:
          - locator: opaque string YOU define, used to re-fetch later
          - title: human-readable title
          - text: full plaintext content (harness handles chunking)
          - metadata: optional dict
        """
        for item in your_format_reader(self._config):
            yield Document(
                locator=item.unique_id,   # you define the format
                title=item.title,
                text=item.get_text(),
                metadata={"source": self._config},
            )

    def retrieve_text(self, locator: str, char_start: int, char_end: int) -> str:
        """
        Re-read a document and return text[char_start:char_end].
        Called at query time. Must be fast (the LLM call is the bottleneck,
        but don't do anything gratuitously slow here).
        """
        return self.retrieve_document(locator)[char_start:char_end]

    def retrieve_document(self, locator: str) -> str:
        """Return the full text of a document. Called by the get_document
        MCP tool and REST endpoint."""
        return your_format_reader.get_by_id(locator)

elif btype == "myformat":
    from potatosearch.backends.my_backend import MyBackend
    backend = MyBackend(
        some_config=entry["config_value"],
        backend_id=backend_id,
    )

{
  "id": "my-data",
  "type": "myformat",
  "description": "My custom data source",
  "config_value": "/path/to/data"
}

cd engine
pipenv run pytest tests/ -v