Code cleanup: lazy clients, lifespan, schemas split, tests and CI

.github/workflows/tests.yml

@@ -0,0 +1,20 @@
+name: tests
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11", "3.12", "3.13"]
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+      - run: pip install -r requirements-dev.txt
+      - run: ruff check .
+      - run: pytest -q

.gitignore

@@ -2,5 +2,6 @@
 .env
 __pycache__
 .DS_Store
-**/.DS_Store
-data/chunk_embeddings.json
+data/chunk_embeddings.json
+.pytest_cache
+.ruff_cache

README.md

@@ -8,7 +8,7 @@ client
   v
 app/main.py  (FastAPI: GET /health, POST /ask, GET /conversations/{session_id})
   |
-  |-- startup: data/knowledge_base.txt
+  |-- startup (lifespan): data/knowledge_base.txt
   |            -> app/knowledge.py      chunk_by_sections()  (split on "# " / "## " headings)
   |            -> app/embedding.py      OpenAI text-embedding-3-small, one call per chunk
   |            -> data/chunk_embeddings.json  (cache, reused while the file's mtime is
@@ -26,23 +26,23 @@ PostgreSQL  (tables: conversations, messages; schema in alembic/versions/001_*.p
 
 Components, by file:
 
-- `app/main.py`: builds the FastAPI app and, at import time, loads and embeds the knowledge base. `POST /ask` takes a `question`, an optional `session_id` (a UUID4 is generated if absent) and an optional `history` list. If `history` is absent or empty, prior messages for that session are loaded from the database. After the LLM answers, the user message and the assistant message are saved. `GET /conversations/{session_id}` returns the stored messages. `GET /health` returns `{"status": "ok"}`.
-- `app/knowledge.py`: `load_system_prompt` reads `prompts/system_prompt.txt`; `chunk_by_sections` starts a new chunk at every line beginning with `# ` or `## ` (`### ` headings stay inside their parent chunk). The sample knowledge base yields 14 chunks. `load_knowledge_base` and `chunk_text` in the same file are not called anywhere.
-- `app/embedding.py`: reads `data/knowledge_base.txt`, calls `text-embedding-3-small`, computes the dot product in plain Python, and caches chunk embeddings in `data/chunk_embeddings.json`. The cache is reused only if the knowledge file's modification time is unchanged; a change to the chunking function alone does not invalidate it.
+- `app/main.py`: builds the FastAPI app and, at startup (FastAPI lifespan), loads and embeds the knowledge base. `POST /ask` takes a `question`, an optional `session_id` (a UUID4 is generated if absent) and an optional `history` list. If `history` is absent or empty, prior messages for that session are loaded from the database. After the LLM answers, the user message and the assistant message are saved. `GET /conversations/{session_id}` returns the stored messages. `GET /health` returns `{"status": "ok"}`.
+- `app/knowledge.py`: `load_system_prompt` reads `prompts/system_prompt.txt`; `chunk_by_sections` starts a new chunk at every line beginning with `# ` or `## ` (`### ` headings stay inside their parent chunk). The sample knowledge base yields 14 chunks.
+- `app/embedding.py`: reads `data/knowledge_base.txt`, calls `text-embedding-3-small` (the OpenAI client is created on first use), computes the dot product in plain Python, and caches chunk embeddings in `data/chunk_embeddings.json`. The cache is reused only if the knowledge file's modification time is unchanged; a change to the chunking function alone does not invalidate it.
 - `app/services/rag_service.py`: retrieval. Returns the concatenated top-3 chunks as context and a `sources` list with each chunk's score and its first 200 characters.
 - `app/llm.py`: sends two system messages (the prompt file, then `Contexto:\n<chunks>`), then the history turns, then the question, to `gpt-4o-mini`.
-- `app/models.py`: the Pydantic request models (`QuestionRequest`, `Message`) and the SQLAlchemy ORM models (`Conversation`, `MessageDB`) live in the same file.
-- `app/services/database_service.py`: `DatabaseService`, a module-level singleton created at import. It opens a session per operation. The API uses two of its methods, `save_message_to_conversation` and `get_conversation_history`; `create_conversation`, `get_conversation` and `add_message` are defined but not called.
-- `alembic/`, `alembic.ini`: one migration (`001`) creating `conversations` and `messages`. `alembic/env.py` reads the connection string from `alembic.ini`, not from `.env`.
-- `init_db.py`: alternative to Alembic; runs `Base.metadata.create_all` with `DATABASE_URL` from `.env` and does not record a migration version.
+- `app/schemas.py`: the Pydantic request models (`QuestionRequest`, `Message`). `app/models.py`: the SQLAlchemy ORM models (`Conversation`, `MessageDB`).
+- `app/services/database_service.py`: `DatabaseService`, a module-level singleton created at import (the engine only connects on first use). It opens a session per operation and has two methods, `save_message_to_conversation` and `get_conversation_history`.
+- `alembic/`, `alembic.ini`: one migration (`001`) creating `conversations` and `messages`. `alembic/env.py` uses `DATABASE_URL` when it is set (importing `app` loads `.env`) and falls back to the placeholder in `alembic.ini` otherwise.
+- `tests/`: chunking, ranking, cache behaviour and the three endpoints, with the OpenAI calls and the database replaced by fakes. `.github/workflows/tests.yml` runs `ruff` and `pytest` on every push.
 - `prompts/system_prompt.txt`: the assistant persona (a support agent for a fictional company "Acme"), in Brazilian Portuguese, instructing the model to answer only from the supplied context.
 - `data/knowledge_base.txt`: the sample knowledge base (plans, billing, support policies of "Acme"), in Portuguese, about 300 lines.
 
 Retrieval happens in process, against the list of chunk embeddings held in memory since startup. PostgreSQL stores chat history only; there is no vector extension and no vector index.
 
 ## Example
 
-The example below is illustrative, not captured from a live run (the app needs a valid OpenAI key even to start: the client is constructed at import, and without a cache the embeddings are generated at import). The request and response fields match the code; the answer text and scores are placeholders. The previews are the first characters of real chunks of `data/knowledge_base.txt`.
+The example below is illustrative, not captured from a live run (without a valid OpenAI key the app cannot finish starting: the embeddings are generated at startup when no cache exists). The request and response fields match the code; the answer text and scores are placeholders. The previews are the first characters of real chunks of `data/knowledge_base.txt`.
 
 ```bash
 curl -s -X POST http://127.0.0.1:8000/ask \
@@ -71,7 +71,7 @@ Sending the returned `session_id` in the next request continues the same convers
    ```bash
    python3 -m venv .venv
    source .venv/bin/activate
-   pip install -r requirements.txt
+   pip install -r requirements.txt   # or requirements-dev.txt to also get pytest and ruff
    ```
 
 2. PostgreSQL. The migration uses only standard column types and no extensions. Create the database:
@@ -82,14 +82,12 @@ Sending the returned `session_id` in the next request continues the same convers
 
 3. Copy `.env.example` to `.env` and fill in both variables. `OPENAI_API_KEY` is read by the `openai` client. `DATABASE_URL` is a SQLAlchemy URL such as `postgresql://USER:PASSWORD@localhost:5432/rag_chatbot`. If `DATABASE_URL` is missing, `database_service.py` falls back to `postgresql://user:password@localhost/rag_chatbot`.
 
-4. Create the tables. Alembic takes its URL from the `sqlalchemy.url` line in `alembic.ini` (a placeholder value is committed); set it to the same value as `DATABASE_URL`, then:
+4. Create the tables. Alembic reads `DATABASE_URL` from `.env`:
 
    ```bash
    alembic upgrade head
    ```
 
-   `python init_db.py` creates the same tables from the ORM models using `.env` instead, without an `alembic_version` table.
-
 5. Start the API. The first start makes one embeddings call per section of the knowledge base and writes `data/chunk_embeddings.json`; later starts reuse it until `data/knowledge_base.txt` changes.
 
    ```bash
@@ -100,7 +98,7 @@ Sending the returned `session_id` in the next request continues the same convers
 
 ## What is not here
 
-- Automated tests: there is no `tests/` directory.
+- Tests against a real database or the real OpenAI API: the test suite replaces both with fakes.
 - Authentication, rate limiting or CORS configuration on the endpoints.
 - Deployment: no Dockerfile, container configuration or CI.
 - Evaluation of answer quality or retrieval quality.

alembic/env.py

@@ -1,7 +1,7 @@
+import os
 from logging.config import fileConfig
 
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
+from sqlalchemy import engine_from_config, pool
 
 from alembic import context
 
@@ -18,9 +18,14 @@
 # for 'autogenerate' support
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
-from app.models import Base
+from app.models import Base  # importing `app` also loads .env
+
 target_metadata = Base.metadata
 
+# DATABASE_URL (from the environment or .env) takes precedence over alembic.ini.
+if os.getenv("DATABASE_URL"):
+    config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"].replace("%", "%%"))
+
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")

app/__init__.py

@@ -0,0 +1,6 @@
+"""RAG chatbot package. Loading the package reads `.env` so every module
+(and Alembic's env.py) sees OPENAI_API_KEY and DATABASE_URL."""
+
+from dotenv import load_dotenv
+
+load_dotenv()

app/embedding.py

@@ -1,53 +1,44 @@
 import json
+from functools import lru_cache
 from pathlib import Path
+
 from openai import OpenAI
 
-client = OpenAI()
+EMBEDDING_MODEL = "text-embedding-3-small"
+
+
+@lru_cache(maxsize=1)
+def get_client() -> OpenAI:
+    # Created on first use, so importing the app does not require an API key.
+    return OpenAI()
 
 
 def generate_embedding(text: str) -> list[float]:
-    response = client.embeddings.create(
-        model="text-embedding-3-small",
-        input=text,
-    )
+    response = get_client().embeddings.create(model=EMBEDDING_MODEL, input=text)
     return response.data[0].embedding
 
 
 def dot_product(a: list[float], b: list[float]) -> float:
-    return sum(x * y for x, y in zip(a, b))
+    return sum(x * y for x, y in zip(a, b, strict=True))
 
 
 def build_chunk_embeddings(chunks: list[str]) -> list[dict]:
-    chunk_embeddings = []
-
-    for chunk in chunks:
-        emb = generate_embedding(chunk)
-        chunk_embeddings.append({
-            "text": chunk,
-            "embedding": emb
-        })
-
-    return chunk_embeddings
+    return [{"text": chunk, "embedding": generate_embedding(chunk)} for chunk in chunks]
 
 
-def rank_chunks_by_similarity(query_embedding: list[float], chunk_embeddings: list[dict]) -> list[dict]:
-    scores = []
-
-    for item in chunk_embeddings:
-        score = dot_product(query_embedding, item["embedding"])
-        scores.append({
-            "text": item["text"],
-            "score": score
-        })
-
+def rank_chunks_by_similarity(
+    query_embedding: list[float], chunk_embeddings: list[dict]
+) -> list[dict]:
+    scores = [
+        {"text": item["text"], "score": dot_product(query_embedding, item["embedding"])}
+        for item in chunk_embeddings
+    ]
     scores.sort(key=lambda x: x["score"], reverse=True)
     return scores
 
 
 def load_or_create_chunk_embeddings(
-    knowledge_path: str,
-    cache_path: str,
-    chunking_function
+    knowledge_path: str, cache_path: str, chunking_function
 ) -> list[dict]:
     knowledge_file = Path(knowledge_path)
     cache_file = Path(cache_path)
@@ -56,7 +47,6 @@ def load_or_create_chunk_embeddings(
 
     if cache_file.exists():
         cache_data = json.loads(cache_file.read_text(encoding="utf-8"))
-
         if cache_data["last_modified"] == current_mtime:
             return cache_data["chunk_embeddings"]
 
@@ -68,12 +58,8 @@ def load_or_create_chunk_embeddings(
     cache_data = {
         "last_modified": current_mtime,
         "chunks": chunks,
-        "chunk_embeddings": chunk_embeddings
+        "chunk_embeddings": chunk_embeddings,
     }
+    cache_file.write_text(json.dumps(cache_data, ensure_ascii=False, indent=2), encoding="utf-8")
 
-    cache_file.write_text(
-        json.dumps(cache_data, ensure_ascii=False, indent=2),
-        encoding="utf-8"
-    )
-
-    return chunk_embeddings
+    return chunk_embeddings

app/knowledge.py

@@ -1,30 +1,14 @@
 from pathlib import Path
 
 
-def load_knowledge_base() -> str:
-    base_path = Path(__file__).resolve().parent.parent
-    file_path = base_path / "data" / "knowledge_base.txt"
-    return file_path.read_text(encoding="utf-8")
-
-
 def load_system_prompt() -> str:
     base_path = Path(__file__).resolve().parent.parent
     file_path = base_path / "prompts" / "system_prompt.txt"
     return file_path.read_text(encoding="utf-8")
 
 
-def chunk_text(text: str, chunk_size: int = 500) -> list[str]:
-    chunks = []
-    start = 0
-
-    while start < len(text):
-        end = start + chunk_size
-        chunks.append(text[start:end])
-        start = end
-
-    return chunks
-
 def chunk_by_sections(text: str) -> list[str]:
+    """Split Markdown text into chunks, starting a new chunk at each `# ` or `## ` heading."""
     chunks = []
     current_chunk = ""
 
@@ -41,4 +25,4 @@ def chunk_by_sections(text: str) -> list[str]:
     if current_chunk:
         chunks.append(current_chunk.strip())
 
-    return chunks
+    return chunks

app/llm.py

@@ -1,41 +1,31 @@
+from functools import lru_cache
+
 from openai import OpenAI
-from typing import List, Dict
 
-client = OpenAI()
+CHAT_MODEL = "gpt-4o-mini"
+
 
+@lru_cache(maxsize=1)
+def get_client() -> OpenAI:
+    # Created on first use, so importing the app does not require an API key.
+    return OpenAI()
 
-def ask_llm(question: str, context: str, system_prompt: str, history: List[Dict[str, str]] = None) -> str:
-    if history is None:
-        history = []
-
+
+def ask_llm(
+    question: str,
+    context: str,
+    system_prompt: str,
+    history: list[dict[str, str]] | None = None,
+) -> str:
     messages = [
-        {
-            "role": "system",
-            "content": system_prompt,
-        },
-        {
-            "role": "system",
-            "content": f"Contexto:\n{context}",
-        },
+        {"role": "system", "content": system_prompt},
+        {"role": "system", "content": f"Contexto:\n{context}"},
     ]
-
-    # Add history messages
-    for msg in history:
-        messages.append({
-            "role": msg["role"],
-            "content": msg["content"]
-        })
-
-    # Add current question
-    messages.append({
-        "role": "user",
-        "content": question,
-    })
-
-    response = client.chat.completions.create(
-        model="gpt-4o-mini",
-        messages=messages,
-    )
+    for msg in history or []:
+        messages.append({"role": msg["role"], "content": msg["content"]})
+    messages.append({"role": "user", "content": question})
+
+    response = get_client().chat.completions.create(model=CHAT_MODEL, messages=messages)
 
     answer = response.choices[0].message.content
-    return answer or "Não foi possível gerar uma resposta."
+    return answer or "Não foi possível gerar uma resposta."