الخطوة 3 في RAG: إعداد pgvector

تضيف pgvector البحث المتجهي إلى PostgreSQL. وبدلًا من SQLite، سنستخدم PostgreSQL مع امتداد pgvector.

لماذا PostgreSQL + pgvector؟

الخطوة 1: تشغيل PostgreSQL باستخدام Docker Compose

أنشئ advanced/docker-compose.rag.yml:

services:
  db:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: aiproject
      POSTGRES_PASSWORD: aiproject
      POSTGRES_DB: aiproject
    ports:
      - "5432:5432"
    volumes:
      - pgvector_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U aiproject"]
      interval: 5s
      timeout: 5s
      retries: 5

  app:
    build: ../backend
    ports:
      - "5000:5000"
    volumes:
      - ../backend:/app
      - sqlite_data:/app/instance
    env_file:
      - ../backend/app/.env
    environment:
      DATABASE_URL: postgresql://aiproject:aiproject@db:5432/aiproject
    depends_on:
      db:
        condition: service_healthy

volumes:
  pgvector_data:
  sqlite_data:

يضيف هذا الملف خدمة PostgreSQL إلى جانب تطبيق Flask لديك. وتتصل خدمة app بقاعدة البيانات عبر متغير البيئة DATABASE_URL.

شغّل الخدمتين:

cd aiproject/advanced
docker compose -f docker-compose.rag.yml up

الخطوة 2: تثبيت التبعيات

pip install psycopg2-binary sentence-transformers

الخطوة 3: إضافة نموذج المستند

أنشئ نموذجًا جديدًا لتخزين أجزاء المستندات وتضميناتها. أضف إلى backend/app/models.py:

from pgvector.sqlalchemy import Vector


class Document(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(500), nullable=False)
    content = db.Column(db.Text, nullable=False)
    source = db.Column(db.String(500), default="")
    category = db.Column(db.String(100), default="general")
    embedding = db.Column(Vector(384), nullable=False)

    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "content": self.content[:200],
            "source": self.source,
            "category": self.category,
        }

يخزّن عمود Vector(384) التضمينات، وسيبحث المعامل <=> حسب مسافة جيب التمام.

أضف أيضًا pgvector إلى requirements.txt:

pgvector
psycopg2-binary
sentence-transformers

الخطوة 4: إنشاء جداول قاعدة البيانات

يلزم إنشاء جدول التضمينات. شغّل التطبيق مرة واحدة بعد ربط PostgreSQL:

cd aiproject
docker compose -f advanced/docker-compose.rag.yml up --build

سينشئ التطبيق الجداول تلقائيًا (عبر db.create_all())، ويمكنك التحقق من ذلك:

docker compose -f advanced/docker-compose.rag.yml exec db psql -U aiproject -c "\dt"

ينبغي أن ترى جداول من بينها جدول document الذي يضم عمود embedding.

الخطوة 5: اختبار الاتصال والنموذج

شغّل هذا الفحص السريع:

# test_pgvector.py
from app import create_app, db
from app.models import Document

app = create_app()
with app.app_context():
    db.create_all()
    count = Document.query.count()
    print(f"Connection OK. Documents in DB: {count}")

شغّله:

DATABASE_URL="postgresql://aiproject:aiproject@localhost:5432/aiproject" python test_pgvector.py
# → Connection OK. Documents in DB: 0

اكتب اختبارًا آليًا

أضف اختبارًا يتحقق من أن نموذج Document يعمل مع pgvector. أنشئ advanced/rag/tests/test_knowledge_base.py:

import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from sample_data import DOCUMENTS
from knowledge_base import chunk_text, get_all_chunks


def test_chunk_text_creates_chunks():
    text = "word " * 500
    chunks = chunk_text(text, chunk_size=100, overlap=20)
    assert len(chunks) >= 5
    assert all(isinstance(c, str) for c in chunks)


def test_chunk_text_short_text():
    text = "short text"
    chunks = chunk_text(text, chunk_size=100, overlap=20)
    assert len(chunks) == 1
    assert chunks[0] == "short text"


def test_get_all_chunks_returns_list():
    chunks = get_all_chunks(DOCUMENTS[:3])
    assert len(chunks) > 0
    assert all("title" in c for c in chunks)
    assert all("content" in c for c in chunks)

شغّلها:

cd aiproject
python -m pytest advanced/rag/tests/ -v

تعمل هذه الاختبارات دون قاعدة بيانات — فهي تختبر منطق التقطيع بمعزل عن بقية الأجزاء.

اكتب اختبارًا سريعًا:

# test_pgvector.py
from app import create_app, db
from app.models import Document

app = create_app()
with app.app_context():
    db.create_all()
    count = Document.query.count()
    print(f"Connection OK. Documents in DB: {count}")

شغّله:

DATABASE_URL="postgresql://aiproject:aiproject@localhost:5432/aiproject" python test_pgvector.py
# → Connection OK. Documents in DB: 0

كيف يعمل

┌─────────────────────────────────────────────────────────┐
│                    pgvector                              │
│                                                         │
│  ┌─────────────────────────────────────────────────┐    │
│  │  document table                                  │    │
│  │  id │ title │ content │ source │ embedding (384) │    │
│  ├─────┼───────┼─────────┼────────┼─────────────────┤    │
│  │  1  │ Tut…  │ Clark…  │ url…   │ [0.04, -0.13,…] │    │
│  │  2  │ Fin…  │ Reso…   │ url…   │ [0.21, 0.07,…]  │    │
│  │  3  │ Dis…  │ The…    │ url…   │ [-0.09, 0.33,…] │    │
│  └─────────────────────────────────────────────────┘    │
│                                                         │
│  Query: SELECT * FROM document                          │
│         ORDER BY embedding <=> :query_vec               │
│         LIMIT 5                                         │
└─────────────────────────────────────────────────────────┘