289 lines
12 KiB
Python
289 lines
12 KiB
Python
"""ذخیره‌ساز برداری pgvector — جایگزین FAISS در مهاجرت به pgvector.
|
||
|
||
بردارها در یک دیتابیس جدا (aramesh_chatbot) روی همان نمونه Postgres سایت
|
||
نگهداری می‌شوند تا دیتابیس سایت فقط‌خواندنی بماند. همگام‌سازی افزایشی است:
|
||
فقط سطرهای جدید/تغییریافته امبد می‌شوند (کلید: doc_id پایدار + hash محتوا).
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
from urllib.parse import urlsplit, urlunsplit
|
||
|
||
from .config import settings
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
def _strip_driver(dsn: str) -> str:
|
||
return dsn.replace("+psycopg", "").replace("+asyncpg", "")
|
||
|
||
|
||
def chatbot_dsn() -> str:
|
||
"""DSN دیتابیس اختصاصی چت‌بات — از DATABASE_URL با تعویض نام دیتابیس ساخته می‌شود."""
|
||
parts = urlsplit(_strip_driver(settings.database_url))
|
||
return urlunsplit(parts._replace(path=f"/{settings.chatbot_db_name}"))
|
||
|
||
|
||
def _admin_dsn() -> str:
|
||
"""DSN دیتابیس postgres برای ساخت دیتابیس چت‌بات (فقط یک‌بار)."""
|
||
parts = urlsplit(_strip_driver(settings.database_url))
|
||
return urlunsplit(parts._replace(path="/postgres"))
|
||
|
||
|
||
def ensure_schema() -> None:
|
||
"""ساخت دیتابیس و جدول بردارها — idempotent و در هر شروع اجرا می‌شود."""
|
||
import psycopg
|
||
from psycopg import errors as pg_errors
|
||
|
||
try:
|
||
with psycopg.connect(_admin_dsn(), autocommit=True, connect_timeout=10) as conn:
|
||
try:
|
||
conn.execute(f'CREATE DATABASE "{settings.chatbot_db_name}"')
|
||
log.info("Created chatbot database '%s'", settings.chatbot_db_name)
|
||
except pg_errors.DuplicateDatabase:
|
||
pass
|
||
except Exception as e:
|
||
raise RuntimeError(f"Cannot reach Postgres to ensure chatbot database: {e}") from e
|
||
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=10) as conn:
|
||
conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
|
||
conn.execute(
|
||
f"""
|
||
CREATE TABLE IF NOT EXISTS chatbot_chunks (
|
||
doc_id TEXT PRIMARY KEY,
|
||
content_hash TEXT NOT NULL,
|
||
model TEXT NOT NULL,
|
||
content JSONB NOT NULL,
|
||
embedding vector({settings.embedding_dim}) NOT NULL
|
||
)
|
||
"""
|
||
)
|
||
conn.execute(
|
||
"""
|
||
CREATE INDEX IF NOT EXISTS chatbot_chunks_embedding_hnsw
|
||
ON chatbot_chunks USING hnsw (embedding vector_cosine_ops)
|
||
"""
|
||
)
|
||
# گزارش گفتگوها — حلقه بازخورد برای کشف پرسش‌های بی‌پاسخ
|
||
conn.execute(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS chat_logs (
|
||
id BIGSERIAL PRIMARY KEY,
|
||
ts TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
thread_id TEXT NOT NULL,
|
||
question TEXT NOT NULL,
|
||
answer TEXT NOT NULL,
|
||
n_docs INTEGER NOT NULL DEFAULT 0,
|
||
is_fallback BOOLEAN NOT NULL DEFAULT FALSE,
|
||
latency_ms INTEGER NOT NULL DEFAULT 0
|
||
)
|
||
"""
|
||
)
|
||
conn.execute("CREATE INDEX IF NOT EXISTS chat_logs_ts_idx ON chat_logs (ts DESC)")
|
||
|
||
|
||
def sync_chunks(chunks: list[dict], force: bool = False) -> dict:
|
||
"""همگام‌سازی افزایشی قطعات با جدول pgvector.
|
||
|
||
- سطرهایی که دیگر وجود ندارند حذف می‌شوند
|
||
- سطرهای جدید یا تغییر یافته (hash یا مدل امبدینگ عوض شده) امبد و upsert می‌شوند
|
||
- force=True همه سطرها را از نو امبد می‌کند
|
||
خروجی: آمار عملیات برای لاگ.
|
||
"""
|
||
import psycopg
|
||
from psycopg.rows import dict_row
|
||
from psycopg.types.json import Jsonb
|
||
|
||
from .embeddings import get_embeddings
|
||
|
||
model = settings.embedding_model
|
||
wanted = {c["doc_id"]: c for c in chunks}
|
||
|
||
# row_factory لازم است: پایین‌تر row["model"]/row["doc_id"] خوانده می‌شود و
|
||
# پیش‌فرض psycopg3 تاپل برمی‌گرداند (خطای tuple indices must be integers).
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=10,
|
||
row_factory=dict_row) as conn:
|
||
existing: dict[str, str] = {}
|
||
if not force:
|
||
for row in conn.execute("SELECT doc_id, content_hash, model FROM chatbot_chunks").fetchall():
|
||
if row["model"] == model:
|
||
existing[row["doc_id"]] = row["content_hash"]
|
||
else:
|
||
conn.execute("DELETE FROM chatbot_chunks")
|
||
|
||
to_delete = [doc_id for doc_id in existing if doc_id not in wanted]
|
||
if to_delete:
|
||
conn.execute(
|
||
"DELETE FROM chatbot_chunks WHERE doc_id = ANY(%s)",
|
||
[to_delete],
|
||
)
|
||
|
||
to_upsert = [
|
||
c for c in chunks
|
||
if existing.get(c["doc_id"]) != c["content_hash"]
|
||
]
|
||
inserted = updated = 0
|
||
|
||
if to_upsert:
|
||
vectors = get_embeddings().embed_documents([c["text"] for c in to_upsert])
|
||
for chunk, vec in zip(to_upsert, vectors):
|
||
literal = "[" + ",".join(f"{x:.7f}" for x in vec) + "]"
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO chatbot_chunks (doc_id, content_hash, model, content, embedding)
|
||
VALUES (%s, %s, %s, %s, %s::vector)
|
||
ON CONFLICT (doc_id) DO UPDATE SET
|
||
content_hash = EXCLUDED.content_hash,
|
||
model = EXCLUDED.model,
|
||
content = EXCLUDED.content,
|
||
embedding = EXCLUDED.embedding
|
||
""",
|
||
(chunk["doc_id"], chunk["content_hash"], model, Jsonb(chunk), literal),
|
||
)
|
||
if chunk["doc_id"] in existing:
|
||
updated += 1
|
||
else:
|
||
inserted += 1
|
||
|
||
stats = {"inserted": inserted, "updated": updated, "deleted": len(to_delete),
|
||
"unchanged": len(wanted) - len(to_upsert)}
|
||
log.info("Vector sync: %s", stats)
|
||
return stats
|
||
|
||
|
||
def fetch_all_chunks() -> list[dict]:
|
||
"""همه قطعات ذخیره‌شده (برای ساخت BM25 در حافظه)."""
|
||
import psycopg
|
||
from psycopg.rows import dict_row
|
||
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=10,
|
||
row_factory=dict_row) as conn:
|
||
rows = conn.execute(
|
||
"SELECT doc_id, content_hash, content FROM chatbot_chunks ORDER BY doc_id"
|
||
).fetchall()
|
||
return [
|
||
{
|
||
"doc_id": r["doc_id"],
|
||
"content_hash": r["content_hash"],
|
||
**(r["content"] if isinstance(r["content"], dict) else json.loads(r["content"])),
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
def vector_count() -> int:
|
||
"""تعداد بردارهای ذخیره‌شده — برای healthz (خطا → ۰)."""
|
||
try:
|
||
import psycopg
|
||
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=5) as conn:
|
||
return int(conn.execute("SELECT count(*) FROM chatbot_chunks").fetchone()[0])
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
def log_chat(thread_id: str, question: str, answer: str, n_docs: int,
|
||
is_fallback: bool, latency_ms: int) -> None:
|
||
"""ثبت یک گفتگو در chat_logs — هرگز خطا نمی‌دهد (لاگ نباید جریان پاسخ را بشکند)."""
|
||
try:
|
||
import psycopg
|
||
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=5) as conn:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO chat_logs (thread_id, question, answer, n_docs, is_fallback, latency_ms)
|
||
VALUES (%s, %s, %s, %s, %s, %s)
|
||
""",
|
||
(thread_id, question, answer[:4000], n_docs, is_fallback, latency_ms),
|
||
)
|
||
except Exception as e:
|
||
log.warning("chat_logs insert failed: %s", e)
|
||
|
||
|
||
def chat_stats(days: int = 7) -> dict:
|
||
"""آمار تجمعی گفتگوها — بدون متن پرسش‌ها (متن‌ها فقط با توکن در /chat/stats)."""
|
||
try:
|
||
import psycopg
|
||
|
||
with psycopg.connect(chatbot_dsn(), autocommit=True, connect_timeout=5) as conn:
|
||
total = int(conn.execute("SELECT count(*) FROM chat_logs").fetchone()[0])
|
||
row = conn.execute(
|
||
"""
|
||
SELECT count(*), count(*) FILTER (WHERE is_fallback)
|
||
FROM chat_logs WHERE ts > now() - (%s || ' days')::interval
|
||
""",
|
||
(str(days),),
|
||
).fetchone()
|
||
recent_fallbacks: list[str] = []
|
||
if row and row[1]:
|
||
recent_fallbacks = [
|
||
r[0] for r in conn.execute(
|
||
"""
|
||
SELECT question FROM chat_logs
|
||
WHERE is_fallback AND ts > now() - (%s || ' days')::interval
|
||
ORDER BY ts DESC LIMIT 20
|
||
""",
|
||
(str(days),),
|
||
).fetchall()
|
||
]
|
||
return {
|
||
"total_logged": total,
|
||
"last_7d": int(row[0]) if row else 0,
|
||
"fallback_7d": int(row[1]) if row else 0,
|
||
"recent_fallback_questions": recent_fallbacks,
|
||
}
|
||
except Exception as e:
|
||
log.warning("chat_stats failed: %s", e)
|
||
return {"total_logged": 0, "last_7d": 0, "fallback_7d": 0, "recent_fallback_questions": []}
|
||
|
||
|
||
def dense_search(query: str, top_k: int) -> list[tuple[dict, float]]:
|
||
"""جست‌وجوی معنایی SQL با فاصله کسینوسی — خروجی: (سند، شباهت).
|
||
|
||
خطا (مثلاً دیتابیس هنوز آماده نیست) → فهرست خالی تا بازیابی فقط با BM25 ادامه یابد.
|
||
"""
|
||
try:
|
||
import psycopg
|
||
from psycopg.rows import dict_row as _dict_row
|
||
|
||
from .embeddings import get_embeddings
|
||
|
||
vec = get_embeddings().embed_query(query)
|
||
literal = "[" + ",".join(f"{x:.7f}" for x in vec) + "]"
|
||
with psycopg.connect(chatbot_dsn(), row_factory=_dict_row, connect_timeout=10) as conn:
|
||
rows = conn.execute(
|
||
f"""
|
||
SELECT doc_id, content, 1 - (embedding <=> %s::vector) AS sim
|
||
FROM chatbot_chunks
|
||
ORDER BY embedding <=> %s::vector
|
||
LIMIT %s
|
||
""",
|
||
(literal, literal, top_k),
|
||
).fetchall()
|
||
return [(_row_to_doc(r), float(r["sim"])) for r in rows]
|
||
except Exception as e:
|
||
log.warning("Dense search failed (%s) — falling back to keyword only", e)
|
||
return []
|
||
|
||
|
||
def _row_to_doc(row: Any) -> dict:
|
||
content = row["content"] if isinstance(row["content"], dict) else json.loads(row["content"])
|
||
return {
|
||
"doc_id": row["doc_id"],
|
||
"text": content.get("text", ""),
|
||
"type": content.get("type", ""),
|
||
"title": content.get("title", ""),
|
||
"url": content.get("url", ""),
|
||
}
|
||
|
||
|
||
def chunk_content_hash(chunk: dict) -> str:
|
||
"""هش محتوای قطعه — تغییر متن یا متادیتا یعنی سطر باید به‌روز شود."""
|
||
payload = json.dumps(
|
||
{k: chunk[k] for k in ("text", "type", "title", "url")},
|
||
ensure_ascii=False, sort_keys=True,
|
||
)
|
||
return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]
|