AI semantic search

# DHTMLX Gantt - Semantic Search Demo

AI-powered semantic search inside a [DHTMLX Gantt](https://dhtmlx.com/docs/products/dhtmlxGantt/) chart. Type natural-language queries and find relevant tasks by meaning, not just keywords.

For a step-by-step implementation guide, see the [Semantic Search tutorial](https://docs.dhtmlx.com/gantt/integrations/ai-tools/semantic-search/) in the DHTMLX Gantt documentation.

## Quick Start

```bash
docker compose up --build
```

Then open **http://localhost:11436** in your browser.

On first launch, the Ollama container downloads the embedding model and warms it up before semantic search becomes available. This can take a little while depending on your machine and network connection.

The sample uses Ollama as a local embedding runtime for simplicity and reproducibility.

## How It Works

1. The Gantt chart loads about 60 tasks from the backend.
2. When you search, the backend generates an embedding for your query using a local model (`all-minilm` via [Ollama](https://ollama.com/)).
3. The backend compares the query embedding with stored task embeddings and returns matches sorted by relevance score. Embeddings for the initial dataset are precomputed, while new or updated tasks are embedded during runtime.
4. Matching tasks are highlighted in the Gantt chart and non-matches are dimmed.

All processing runs locally. No external AI API is called while you use the demo.

## Runtime Notes

- CRUD changes are stored in memory only and are reset when the backend container restarts.
- Semantic search may be temporarily unavailable during initial model download or model warm-up.
- To test with a different embedding model, change `MODEL_NAME` in `backend/main.py`, restart the backend, and call `GET /api/embeddings/recalculate` to regenerate all task embeddings.

## Example Queries

- `"authentication and security"` - finds tasks about auth modules, security assessment, and social login
- `"deployment pipeline"` - finds CI/CD configuration, production setup, and monitoring tasks
- `"user interface"` - finds UI prototyping, responsive design, and onboarding flow tasks
- `"database and storage"` - finds schema design, ORM setup, and data migration tasks

## Architecture

```text
Browser (port 11436)
  -> nginx (reverse proxy)
     -> /          -> static frontend (index.html)
     -> /api/*              -> FastAPI backend (port 11435)
     -> /api/embeddings/*   -> FastAPI backend (port 11435)
     -> /search             -> FastAPI backend (port 11435)
                               -> Ollama (port 11434)
```

| Service  | Port  | Description |
|----------|-------|-------------|
| Frontend | 11436 | nginx serving the Gantt UI |
| Backend  | 11435 | FastAPI - CRUD plus semantic search |
| Ollama   | 11434 | Local embedding runtime used by the demo |

## Project Structure

```text
frontend/
  index.html       # HTML shell and search toolbar
  main.js          # Gantt init, search logic, DataProcessor
  styles.css       # Dark theme, search highlight styling
  nginx.conf       # Reverse proxy config
  Dockerfile
backend/
  main.py          # FastAPI server
  data.json        # Task data plus precomputed embeddings
  requirements.txt
  Dockerfile
ollama/
  init.sh          # Model pull script
  Dockerfile
docker-compose.yml
```

## Requirements

- [Docker](https://docs.docker.com/get-docker/) and Docker Compose

## License

Source code in this repo is released under the **MIT License**.

**DHTMLX Gantt** is a commercial library - use under a valid [DHTMLX license](https://dhtmlx.com/docs/products/licenses.shtml) or evaluation agreement.

## Useful links

- [Semantic Search Tutorial](https://docs.dhtmlx.com/gantt/integrations/ai-tools/semantic-search/) - step-by-step implementation guide
- [DHTMLX Gantt Product Page](https://dhtmlx.com/docs/products/dhtmlxGantt/)
- [DHTMLX Gantt Documentation](https://docs.dhtmlx.com/gantt/)
- [Ollama](https://ollama.com/) - local embedding runtime used by this demo
- [all-minilm Model](https://ollama.com/library/all-minilm) - embedding model used for semantic search
- [DHTMLX Technical Support Forum](https://forum.dhtmlx.com/)

services:

  ollama:
    build: ./ollama
    container_name: ollama_server
    ports:
      - "11434:11434"
    volumes:
      - ollama:/root/.ollama  

  backend:
    build: ./backend
    container_name: backend_server
    ports:
      - "11435:11435"
    environment:
      - OLLAMA_HOST=http://ollama:11434
    depends_on:
      - ollama
    command:
      - python
      - main.py

  frontend:
    build: ./frontend
    container_name: frontend_server
    ports:
      - "11436:80"
    depends_on:
      - backend

volumes:

  ollama:

import fastapi
import fastapi.middleware.cors
import html
import json
import logging
import ollama
import pydantic
import uvicorn

TaskId = str | int
LinkId = str | int

class ModelTaskAPI(pydantic.BaseModel):
    id: TaskId
    text: str
    description: str
    parent: TaskId
    progress: float
    duration: int
    start_date: str
    end_date: str

class ModelLinkAPI(pydantic.BaseModel):
    id: LinkId
    source: TaskId
    target: TaskId
    type: str

class SearchRequest(pydantic.BaseModel):
    query: str

class SearchResult(pydantic.BaseModel):
    id: TaskId
    score: float



class Application:

    MODEL_NAME = "all-minilm"
    SIMILARITY_THRESHOLD = 0.4

    @staticmethod
    def sanitize_task(task: ModelTaskAPI) -> ModelTaskAPI:
        return task.model_copy(update = {
            "text": html.escape(task.text),
            "description": html.escape(task.description),
        })

    @staticmethod
    def get_indexable_text(task: ModelTaskAPI) -> str:
        return f"{task.text}\n{task.description}"

    @staticmethod
    def get_embedding(text: str) -> list[float]:
        response = ollama.embed(
            model = Application.MODEL_NAME,
            input = text,
            truncate = True
        )
        return response.embeddings[0]

    def __init__(self):

        with open("data.json", "r", encoding = "utf-8") as file:

            data = json.load(file)
            self.document_storage: dict[TaskId, ModelTaskAPI] = {
                record["id"] : ModelTaskAPI(
                    id = record["id"],
                    text = record["text"],
                    description = record["description"],
                    parent = record["parent"],
                    progress = record["progress"],
                    duration = record["duration"],
                    start_date = record["start_date"],
                    end_date = record["end_date"]
                )
                for record in data["document_storage"].values()
            }
            self.link_storage: dict[LinkId, ModelLinkAPI] = {
                record["id"] : ModelLinkAPI(
                    id = record["id"],
                    source = record["source"],
                    target = record["target"],
                    type = record["type"]
                )
                for record in data["link_storage"].values()
            }
            self.vector_storage: dict[TaskId, list[float]] = {
                data["document_storage"][id]["id"] : record
                for id, record in data["vector_storage"].items()
            }

        self.application = fastapi.FastAPI()
        self.application.add_middleware(
            fastapi.middleware.cors.CORSMiddleware,
            allow_origins = ["*"],
            allow_credentials = True,
            allow_methods = ["*"],
            allow_headers = ["*"],
        )

        @self.application.get("/api/read")
        async def read():

            try:

                return fastapi.responses.JSONResponse(
                    content = fastapi.encoders.jsonable_encoder({
                        "tasks": [
                            {**document.model_dump(), "open": True}
                            for _, document in self.document_storage.items()
                        ],
                        "links": [
                            {**link.model_dump()}
                            for _, link in self.link_storage.items()
                        ]
                    })
                )

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/tasks/create")
        async def tasks_create(request: ModelTaskAPI):

            try:

                request = Application.sanitize_task(request)
                embedding = Application.get_embedding(Application.get_indexable_text(request))

                self.document_storage[request.id] = request
                self.vector_storage[request.id] = embedding

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "created",
                        "tid": request.id
                    }
                )

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/tasks/update")
        async def tasks_update(request: ModelTaskAPI):

            try:

                if request.id not in self.document_storage:
                    raise fastapi.HTTPException(status_code = 404, detail = f"Task '{request.id}' not found")

                request = Application.sanitize_task(request)
                embedding = Application.get_embedding(Application.get_indexable_text(request))

                self.document_storage[request.id] = request
                self.vector_storage[request.id] = embedding

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "updated"
                    }
                )

            except fastapi.HTTPException:
                raise

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/tasks/delete")
        async def tasks_delete(request: ModelTaskAPI):

            try:

                if request.id not in self.document_storage:
                    raise fastapi.HTTPException(status_code = 404, detail = f"Task '{request.id}' not found")

                del self.document_storage[request.id]
                del self.vector_storage[request.id]

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "deleted"
                    }
                )
            
            except fastapi.HTTPException:
                raise

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/links/create")
        async def links_create(request: ModelLinkAPI):

            try:

                self.link_storage[request.id] = request
     
                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "created", 
                        "tid": request.id
                    }
                )

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/links/update")
        async def links_update(request: ModelLinkAPI):

            try:

                if request.id not in self.link_storage:
                    raise fastapi.HTTPException(status_code = 404, detail = f"Link '{request.id}' not found")

                self.link_storage[request.id] = request

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "updated"
                    }
                )

            except fastapi.HTTPException:
                raise

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post("/api/links/delete")
        async def links_delete(request: ModelLinkAPI):

            try:

                if request.id not in self.link_storage:
                    raise fastapi.HTTPException(status_code = 404, detail = f"Link '{request.id}' not found")

                del self.link_storage[request.id]

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "deleted", 
                        "tid": request.id
                    }
                )
            
            except fastapi.HTTPException:
                raise

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        # for testing/demo purposes only - recalculates all embeddings after changing the model
        @self.application.get("/api/embeddings/recalculate")
        async def embeddings_recalculate():

            try:

                for id, task in self.document_storage.items():
                    self.vector_storage[id] = Application.get_embedding(Application.get_indexable_text(task))

                return fastapi.responses.JSONResponse(
                    content = {
                        "action": "recalculated",
                        "count": len(self.vector_storage)
                    }
                )

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        @self.application.post(
            "/search",
            operation_id = "search",
            summary = "Search",
            description = (
                "Receives user's query related to the Gantt tasks. "
                "Performs semantic vector-based search. "
                "Returns results sorted by similarity score (highest first)."
            ),
            response_model = list[SearchResult]
        )
        async def search(request: SearchRequest) -> list[SearchResult]:

            try:
                
                embedding = Application.get_embedding(request.query)

                response = []

                for id, vector in self.vector_storage.items():

                    product = sum(x * y for x, y in zip(vector, embedding))
                    normX = sum(x * y for x, y in zip(vector, vector)) ** 0.5
                    normY = sum(x * y for x, y in zip(embedding, embedding)) ** 0.5
                    similarity = 0.0 if (normX == 0 or normY == 0) else product / (normX * normY)

                    if similarity > Application.SIMILARITY_THRESHOLD:

                        response.append(
                            SearchResult(
                                id = id, 
                                score = round(similarity, 4)
                            )
                        )

                response.sort(key = lambda record: record.score, reverse = True)

                return response

            except Exception as exception:

                logging.exception(str(exception))
                raise fastapi.HTTPException(status_code = 500, detail = str(exception))

        uvicorn.run(self.application, host = "0.0.0.0", port = 11435)



if __name__ == "__main__":

    application = Application()

fastapi==0.135.1
ollama==0.6.1
pydantic==2.12.5
uvicorn==0.41.0