Build a Semantic Search-Powered FAQ Assistant with TiDB and AWS Bedrock

cd semantic-qna
python -m venv .venv
# In PowerShell
.venvScriptsActivate.ps1
# Or in Command Prompt
.venvScriptsactivate.bat

# PowerShell
$Env:AWS_REGION="us-east-1"
$Env:AWS_ACCESS_KEY_ID="YOUR_AWS_KEY"
$Env:AWS_SECRET_ACCESS_KEY="YOUR_AWS_SECRET"
$Env:DATABASE_URL="mysql+pymysql://<user>:<pass>@…"

REM Command Prompt
set AWS_REGION=us-east-1
set AWS_ACCESS_KEY_ID=YOUR_AWS_KEY
set AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET
set DATABASE_URL=mysql+pymysql://<user>:<pass>@...

# 1. Clone the repo
git clone https://github.com/RealChrisSean/semantic-qna.git
cd semantic-qna

# 2. Create a virtual environment (recommended)
python3 -m venv .venv
source .venv/bin/activate   # (.venv) will appear in your shell prompt

# 3. Install the necessary packages
pip install --upgrade pip
pip install -r requirements.txt

AWS_REGION=us-east-1
AWS_ACCESS_KEY_ID=YOUR_AWS_KEY
AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET

**Note: The TiDB Cloud hostname will vary by region. Make sucopy yours from the TiDB Cloud console**
DATABASE_URL=mysql+pymysql://<TIDB_USER>:<TIDB_PASSWORD>@
gateway01.us-west-2.prod.aws.tidbcloud.com:4000/test?
ssl_ca=/etc/ssl/cert.pem&ssl_verify_cert=true&ssl_verify_identity=true

[
  {
    "id": "1",
    "question": "What is your return policy?",
    "answer": "You can return items within 30 days."
  },
  {
    "id": "2",
    "question": "How long does shipping take?",
    "answer": "Standard shipping takes 3–5 business days."
  },
  {
    "id": "3",
    "question": "Do you ship internationally?",
    "answer": "Yes, we ship to over 35 countries."
  },
  {
    "id": "4",
    "question": "How do I track my order?",
    "answer": "Check the tracking link in your confirmation email."
  }
]

import os, json
from pathlib import Path
from dotenv import load_dotenv       # loads our .env file
import boto3                         # AWS SDK for Python
from typing import List
from tidb_vector.integrations import TiDBVectorClient

# Load all variables from .env
load_dotenv()

# Read environment variables, with some defaults
AWS_REGION = os.getenv("AWS_REGION", "us-east-1")
TIDB_CONN_STR  = os.getenv("DATABASE_URL")
VECTOR_DIM = 1024        # Titan-V2 embeddings are always 1024 elements
TABLE_NAME = "faqs"      # name of the vector table we'll create

def bedrock_embed(text: str) -> List[float]:
    """
    Convert a single string into a Titan-V2 embedding (1024 floats).
    """
    # Create a 'bedrock-runtime' client in the correct region
    brt = boto3.client("bedrock-runtime", region_name=AWS_REGION)

    # Prepare the JSON payload. We just pass 'inputText': the text to embed.
    payload = {"inputText": text}

    # The 'invoke_model' call sends the payload to Amazon Titan
    resp = brt.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",   # The specific Titan embedding model
        contentType="application/json",
        accept="application/json",
        body=json.dumps(payload),
    )

    # The response body is a streaming object, so we read and JSON-parse it
    response_body = json.loads(resp["body"].read())

    # Finally, return the list of floats
    return response_body["embeddingsByType"]["float"]

def batch_embed_batch(texts: List[str]) -> List[List[float]]:
    """
    Embed multiple texts in one Titan batch request.
    Returns a list of 1024-float vectors, one per input string.
    """
    brt = boto3.client("bedrock-runtime", region_name=AWS_REGION)
    resp = brt.invoke_model(
        modelId=os.getenv("BEDROCK_MODEL_ID", "amazon.titan-embed-text-v2:0"),
        contentType="application/json",
        accept="application/json",
        body=json.dumps({"inputTextArray": texts}),
    )
    data = json.loads(resp["body"].read())
    try:
        return data["embeddingsByType"]["floatArray"]

def ingest_faqs(file_path: str = FAQ_FILE):
    """Create the table and load FAQs from a JSON file."""
    client = TiDBVectorClient(
        table_name          = TABLE_NAME,
        connection_string   = TIDB_CONN_STR,
        vector_dimension    = VECTOR_DIM,
        drop_existing_table = False,
    )

    # Check if the table already has data using a standalone engine
    engine = create_engine(TIDB_CONN_STR)
    with engine.connect() as conn:
        row = conn.execute(text(f"SELECT 1 FROM {TABLE_NAME} LIMIT 1")).first()
    if row is not None:
        return client

    faq_path = Path(file_path)
    with faq_path.open("r", encoding="utf-8") as f:
        faqs = json.load(f)

    ids   = [row["id"] for row in faqs]
    texts = [row["question"] for row in faqs]
    metas = [{"answer": row["answer"]} for row in faqs]

    # Batch embed all questions at once
    embs = batch_embed_batch(texts)

    client.insert(ids=ids, texts=texts, embeddings=embs, metadatas=metas)
    return client

def query_faq(question: str, client=None):
    """
    1) Embed the user question via bedrock_embed
    2) Query TiDB for the single closest vector
    3) Return that FAQ's question and answer
    """
    # Step 1: Convert user question into a 1024-dim vector
    q_vec = bedrock_embed(question)

    # Step 2: Run a vector search for the top-1 match
    results = client.query(q_vec, k=1)

    # If no results, we return a "not found" structure
    if not results:
        return {"question": None, "answer": None}

    # Otherwise, get the best match (index 0)
    best = results[0]

    # We expect 'text' to hold the question text (some versions might call it 'payload')
    stored_question = getattr(best, "text", None) or getattr(best, "payload", f"<id {best.id}>")

    # 'metadata' is where we stored the FAQ answer
    stored_answer = (best.metadata or {}).get("answer", "")

    return {"question": stored_question, "answer": stored_answer}

def main():
    # Ingest the FAQs, storing them in TiDB. Returns a client object for queries.
    client = ingest_faqs()

    # We'll prompt until the user types 'exit' or 'quit'
    while True:
        user_q = input("n🧐  Ask a question ('exit' to quit): ").strip()
        if user_q.lower() in {"exit", "quit"}:
            break

        result = query_faq(user_q, client)
        if not result["question"]:
            print("🤷  No semantic match found.")
        else:
            print(f"🎯  Closest Q: {result['question']}")
            print(f"💡  Answer:   {result['answer']}")

if __name__ == "__main__":
    main()

python app.py

from pathlib import Path
from pydantic import BaseModel
from app import ingest_faqs, query_faq
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # We'll reuse this client across requests
    global client
    client = ingest_faqs()
    yield

app = FastAPI(lifespan=lifespan)

# Serve our React index.html at the root path
@app.get("/", include_in_schema=False)
async def read_index():
    return FileResponse(Path(__file__).parent / "index.html")

client = None

# A Pydantic model for the incoming JSON from React
class QueryReq(BaseModel):
    question: str

@app.post("/query")
def ask(req: QueryReq):
    # Use the same logic we have in app.py
    res = query_faq(req.question, client)
    if not res["question"]:
        raise HTTPException(404, "No semantic match found")
    return res

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ubpgbjhzoxsni0v0b5f2.png)
def health():
    return {"status": "ok"}

python run_with_bar.py