ПРОВЕРЕНО · 01.10.2026 ОБНОВЕНО · 01.10.2026

GraphRAG, продукционен RAG и оценка с RAGAS

Векторното търсене намира подобен текст. Графът на знанието намира връзките: кой член препраща към кой, кой плаща, какво се изисква. В този урок затваряме Блок 4: строим граф от правни текстове, сглобяваме RAG услуга с кеш и я мерим всяка нощ с RAGAS — на локален модел.

⏱ 4–5 ч Напреднало Блок 4 · Част 2/2 граф на знанието · кеш · оценка

NetworkX (графът)🔒 локално Ollama (моделите)🔒 локално FastAPI · Redis🔒 локално RAGAS (оценка)🔒 локално Сигнал в чат (webhook)🌐 глобален

🔄

ОБНОВЕНО · 01.10.2026 — какво

Оценката е пренаписана за RAGAS 0.4 (сверено в PyPI и в пясъчник): метриките идват от ragas.metrics.collections, моделът-съдия — през llm_factory и OpenAI-съвместимия адрес на Ollama. Старите LangchainLLMWrapper и импортите от ragas.metrics вече са остарели. Открихме капан: RAGAS 0.4.3 не се зарежда с langchain-community 0.4.2 — закрепи по-стара версия. Извличането на граф минава през структуриран изход вместо regex; обхождането вече гледа и входящите връзки; услугата ползва ainvoke, aclose(), ограничен CORS, версия на индекса в ключа на кеша и изрична команда за изчистване. Юридическите примери са сверени: обезщетенията по болничен плаща НОИ (не НЗОК), членовете са 40–41 КСО (не „45–46“); махнати са непроверими еталони („6 месеца“, „направление, здравна книжка“). Историята за „болница, 847 члена, 0.61 → 0.89“ не можахме да потвърдим — махната; „completeness“ не е метрика на RAGAS.

01Какво ще научиш

Кога графът на знанието побеждава обикновения RAG — и кога не си струва.
Как да извлечеш възли и връзки от текст с локален модел и строга схема.
Как да обходиш графа на няколко стъпки и да смесиш резултата с векторното търсене.
Как да сглобиш RAG услуга с FastAPI и кеш в Redis, която не връща остарели отговори.
Какво мерят четирите основни метрики на RAGAS и как да ги пуснеш с локален съдия.
Как да автоматизираш нощна оценка със сигнал, когато качеството падне.

02Преди да започнеш

Минал си Блок 4 · Част 1 (напреднал RAG): там са функциите advanced_retrieve (в reranker_pipeline.py) и answer (в full_rag_pipeline.py). Тук ги внасяме през кратък файл rag_core.py с два реда: from reranker_pipeline import advanced_retrieve as advanced_rag_retrieve и from full_rag_pipeline import answer as advanced_rag_answer.
Python 3.10 или по-нов, виртуална среда, Redis 🔒 локално (или Docker образ redis).
Ollama 🔒 локално с модел за отговори, по-силен модел за съдия и модел за embeddings.

bash · инсталация (версии към 01.10.2026)

python3 -m venv .venv && source .venv/bin/activate
pip install -U networkx langchain-ollama fastapi uvicorn "redis>=5" ragas openai "langchain-community<0.4.2"
# проверено с: ragas 0.4.3 · networkx 3.7 · langchain-ollama 1.1.0 · redis 8.1 · fastapi 0.142
ollama pull qwen3:8b          # отговаря и извлича графа
ollama pull qwen3:14b         # съдия за RAGAS
ollama pull nomic-embed-text  # embeddings

⚠️

Капан при инсталация

RAGAS 0.4.3 (последната версия към 01.10.2026) при зареждане внася модул, който langchain-community 0.4.2 вече няма. Без ограничението <0.4.2 получаваш ModuleNotFoundError: langchain_community.chat_models.vertexai още на import ragas.

03Стъпки

Кога графът побеждава обикновения RAG

Обикновеният RAG реже документите на парчета и търси най-подобните. Това е достатъчно, когато отговорът стои в едно парче. Защо тогава граф? Защото в правото и в процедурите отговорът е разпръснат: един член препраща към друг, институция финансира процедура, документ се изисква от трети член. Графът пази тези връзки изрично.

Тип въпрос	Обикновен RAG	GraphRAG
„Какво урежда чл. 87 ЗЗД?“	✅ Директно съвпадение	✅ Същото + знае свързаните членове
„Кой плаща болничния и по кой член?“	⚠️ Често намира само едното парче	✅ болничен → чл. 40 → работодател; → чл. 41 → НОИ
„Кои членове на ЗЗД уреждат обезщетението при неизпълнение?“	⚠️ Само при съвпадение на думи	✅ неизпълнение → обезщетение → свързаните членове
„Каква е последователността при планов прием в болница?“	❌ Не разбира последователност	✅ Път в графа: направление → прием → клинична пътека

✅

Кога граф

Правни текстове с препратки, процедурни наръчници, области с много взаимни зависимости. Цена: извличането на граф минава всеки откъс през модел — индексирането става много по-бавно, а всяко търсене добавя още една стъпка. Мери забавянето на своя хардуер, преди да го включиш за всички въпроси.

пример · малък граф (учебен)

КСО ──СЪДЪРЖА──→ чл. 40 ──УРЕЖДА──→ болничен лист ←──ИЗПЛАЩА── НОИ
 │                  └── първите 2 работни дни: работодателят, 70%
 └──СЪДЪРЖА──→ чл. 41 ──УРЕЖДА──→ обезщетение 80% ──ИЗЧИСЛЯВА_СЕ_ОТ──→ среднодневно брутно (18 мес.)

НЗОК ──ФИНАНСИРА──→ клинична пътека ──ПОКРИВА──→ хоспитализация

Въпрос „кой плаща болничния?“ → старт: „болничен лист“ → 2 стъпки → чл. 40, чл. 41, НОИ, КСО

💡

Фактите в примера (сверени)

Обезщетенията при временна неработоспособност плаща НОИ, не НЗОК. От 2024 г. работодателят плаща първите два работни дни по 70% (чл. 40, ал. 5 КСО); НОИ плаща 80% от среднодневното брутно за 18 месеца, но не повече от среднодневното нетно (чл. 41 КСО). НЗОК финансира клиничните пътеки. Учебен пример, не правен съвет.

Извличане на граф с локален модел

Моделът чете откъс и връща възли (закон, член, понятие, институция) и връзки между тях. Защо със схема? Старият вариант молеше за JSON и го вадеше с regex — при първата грешна скоба откъсът тихо изчезваше. С with_structured_output моделът попълва Pydantic обект, а грешката се вижда.

Python · graph_rag_builder.py

import networkx as nx
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen3:8b", temperature=0)

# ── Схемата на графа: моделът попълва Pydantic, не свободен JSON ──
class Entity(BaseModel):
    id: str = Field(description="Кратък уникален ключ, напр. KSO_40")
    type: str = Field(description="закон | член | понятие | институция | документ")
    name: str = Field(description="Име за хора, напр. „чл. 40 КСО“")
    description: str = ""

class Relation(BaseModel):
    source: str = Field(description="id на изходния възел")
    target: str = Field(description="id на целевия възел")
    type: str = Field(description="СЪДЪРЖА | УРЕЖДА | ПРЕПРАЩА | ИЗИСКВА | ФИНАНСИРА | ДЕФИНИРА")

class GraphExtraction(BaseModel):
    entities: list[Entity]
    relations: list[Relation]

PROMPT = ChatPromptTemplate.from_messages([
    ("system", "Извлечи знаниев граф от правния текст. Само това, което е в текста — "
               "не добавяй членове и закони по памет."),
    ("human", "{text}"),
])
extractor = PROMPT | llm.with_structured_output(GraphExtraction)

class LegalKnowledgeGraph:
    def __init__(self):
        self.graph = nx.DiGraph()

    def extract_and_add(self, text: str, source: str = "document", max_chars: int = 3000) -> int:
        """Извлича възли и връзки от откъс текст и ги добавя в графа."""
        try:
            data = extractor.invoke({"text": text[:max_chars]})
        except Exception as err:                      # моделът не спази схемата
            print(f"пропуснат откъс от {source}: {err}")
            return 0
        for e in data.entities:
            if e.id in self.graph:                    # вече го има → само добавяме източника
                self.graph.nodes[e.id]["sources"].add(source)
                continue
            self.graph.add_node(e.id, name=e.name, entity_type=e.type,
                                description=e.description, sources={source})
        for r in data.relations:
            if r.source in self.graph and r.target in self.graph:
                self.graph.add_edge(r.source, r.target, relation=r.type)
        return len(data.entities)

    def find_start_nodes(self, query: str, limit: int = 3) -> list[str]:
        """Прост старт: съвпадение по началото на думите (болничен/болничния).
        В production — търсене по embeddings на имената на възлите."""
        stems = {w[:6] for w in query.lower().split() if len(w) > 3}
        scored = []
        for node, d in self.graph.nodes(data=True):
            label = f"{node} {d.get('name', '')}".lower()
            score = sum(s in label for s in stems)
            if score:
                scored.append((score, node))
        return [n for _, n in sorted(scored, reverse=True)[:limit]]

    def get_context_for_query(self, query: str, hops: int = 2) -> list[str]:
        """Събира възлите на до `hops` стъпки от стартовите — в двете посоки."""
        nodes = set()
        for start in self.find_start_nodes(query):
            # undirected=True: иначе от „чл. 40“ не стигаш до закона, който го съдържа
            nodes |= set(nx.ego_graph(self.graph, start, radius=hops, undirected=True))
        lines = []
        for n in nodes:
            d = self.graph.nodes[n]
            rels = [f"{e['relation']} → {self.graph.nodes[v]['name']}"
                    for _, v, e in self.graph.out_edges(n, data=True)]
            line = f"[{d['entity_type']}] {d['name']}"
            if d.get("description"):
                line += f": {d['description']}"
            if rels:
                line += " | връзки: " + "; ".join(rels[:3])
            lines.append(line)
        return lines

# ── Хибрид: векторно търсене (урок 01-04a) + контекст от графа ──
def graph_hybrid_rag(query: str, kg: LegalKnowledgeGraph, retrieve, top_k: int = 5) -> dict:
    vector_docs = retrieve(query, top_k=top_k)        # напр. advanced_rag_retrieve от 01-04a
    graph_context = kg.get_context_for_query(query, hops=2)
    context = "\n\n".join(["## Граф на знанието (структурирано):", *graph_context[:8],
                           "## Текстови откъси:", *[d["text"] for d in vector_docs]])
    return {"context": context, "graph_nodes": len(graph_context),
            "sources": [d.get("metadata", {}) for d in vector_docs]}

⚠️

Моделът може да „допише“ закона

Локалните модели понякога добавят членове, които ги няма в откъса, или бъркат номера. Пази източника на всеки възел (sources) и прави извадкова проверка от човек, преди графът да влезе в работа.

🧭

Кога да минеш на по-голям инструмент

NetworkX държи графа в паметта на процеса — удобно за учене и за малки графи. При голям граф, много потребители или нужда от заявки (Cypher) премини на Neo4j с пакета neo4j-graphrag. Microsoft GraphRAG добавя групиране на общности и „глобално“ търсене по целия корпус, срещу по-скъпо индексиране.

Продукционна RAG услуга с кеш

Продукционният RAG не е само „търси и генерирай“. Нужни са: индексиране, кеш, мониторинг, версии и постоянна оценка. Тук сглобяваме услугата. Защо кеш? Едни и същи въпроси идват многократно, а генерирането е най-скъпата стъпка. Защо версия в ключа? Защото след обновяване на документите старият кеш лъже.

Python · production_rag_api.py

import asyncio, hashlib, json, os, time
from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama

from rag_core import advanced_rag_retrieve                 # от урок 01-04a
from graph_rag_builder import LegalKnowledgeGraph, graph_hybrid_rag

MODEL = os.getenv("RAG_MODEL", "qwen3:8b")
INDEX_VERSION = os.getenv("INDEX_VERSION", "v1")           # вдигни при всяко преиндексиране
knowledge_graph = LegalKnowledgeGraph()                     # в реална система — зареден от диск

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"),
                                     decode_responses=True)
    app.state.llm = ChatOllama(model=MODEL, temperature=0.1)
    yield
    await app.state.redis.aclose()                          # close() е остаряло в redis-py 5+

app = FastAPI(title="RAG API", lifespan=lifespan)
app.add_middleware(CORSMiddleware,                          # само твоите домейни, не "*"
                   allow_origins=os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(","),
                   allow_methods=["GET", "POST"], allow_headers=["Content-Type"])

class RAGRequest(BaseModel):
    question: str
    course_id: str | None = None
    use_graph: bool = False
    top_k: int = 5
    cache_ttl: int = 3600                                   # секунди; 0 = без кеш

class RAGResponse(BaseModel):
    answer: str
    sources: list[dict]
    cached: bool
    latency_ms: int
    graph_nodes_used: int = 0

RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", "Отговаряй САМО от контекста, на български. "
               "Ако отговорът не е в контекста — кажи го направо."),
    ("human", "Контекст:\n{context}\n\nВъпрос: {question}"),
])

def cache_key(req: RAGRequest) -> str:
    raw = f"{req.question}|{req.use_graph}|{req.top_k}"
    return f"rag:{req.course_id or 'all'}:{INDEX_VERSION}:{hashlib.sha256(raw.encode()).hexdigest()[:16]}"

@app.post("/rag/query", response_model=RAGResponse)
async def rag_query(req: RAGRequest, request: Request):
    t0 = time.perf_counter()
    r, llm = request.app.state.redis, request.app.state.llm

    key = cache_key(req) if req.cache_ttl > 0 else None     # 1. кеш
    if key and (hit := await r.get(key)):
        data = json.loads(hit)
        return RAGResponse(**data, cached=True, latency_ms=int((time.perf_counter() - t0) * 1000))

    graph_nodes = 0                                         # 2. търсене (синхронно → в нишка)
    if req.use_graph:
        g = await asyncio.to_thread(graph_hybrid_rag, req.question, knowledge_graph,
                                    advanced_rag_retrieve, req.top_k)
        context, sources, graph_nodes = g["context"], g["sources"], g["graph_nodes"]
    else:
        docs = await asyncio.to_thread(advanced_rag_retrieve, req.question,
                                       course_id=req.course_id, top_k=req.top_k)
        context = "\n\n".join(d["text"] for d in docs)
        sources = [d.get("metadata", {}) for d in docs]

    msg = await (RAG_PROMPT | llm).ainvoke({"context": context, "question": req.question})  # 3. отговор

    data = {"answer": msg.content, "sources": sources, "graph_nodes_used": graph_nodes}
    if key:                                                 # 4. запис в кеша
        await r.setex(key, req.cache_ttl, json.dumps(data, ensure_ascii=False))
    return RAGResponse(**data, cached=False, latency_ms=int((time.perf_counter() - t0) * 1000))

@app.post("/cache/flush")                                   # в production — зад автентикация
async def flush_cache(request: Request, course_id: str | None = None):
    r, deleted = request.app.state.redis, 0
    async for k in r.scan_iter(match=f"rag:{course_id or '*'}:*", count=500):
        deleted += await r.delete(k)
    return {"deleted": deleted}

@app.get("/health")
async def health():
    return {"status": "ok", "model": MODEL, "index_version": INDEX_VERSION}

# uvicorn production_rag_api:app --host 127.0.0.1 --port 8000 --workers 2

✅

Обновена наредба = преиндексиране + изчистване на кеша

Не чакай кешът да изтече сам — дотогава потребителите получават стари отговори. Изтрий старите парчета на документа от векторната база, индексирай новите, после вдигни INDEX_VERSION или извикай /cache/flush?course_id=…. Най-добре — като последна стъпка на автоматизацията, която индексира.

⚠️

Три поправки спрямо стария код

(1) allow_origins=["*"] отваря услугата за всеки сайт — ограничи я. (2) Синхронното chain.invoke в async функция блокира всички заявки, докато моделът мисли — ползвай ainvoke, а бавното търсене пусни в нишка. (3) Ако графът е изграден в паметта, всеки от работниците на uvicorn пази свое копие — зареждай го от файл или от база.

RAGAS: какво мерим

Без измерване не знаеш дали системата се е влошила след нови документи или смяна на модела. RAGAS ползва модел-съдия, който оценява всяка двойка въпрос–отговор. Четири метрики покриват двете половини на RAG — търсенето и генерирането:

Метрика	Какво пита	Какво ѝ трябва	Примерен праг
Faithfulness	Подкрепено ли е всяко твърдение в отговора от намерения контекст? (срещу измислици)	въпрос, отговор, контекст	≥ 0.90
Answer relevancy	Отговаря ли отговорът на зададения въпрос?	въпрос, отговор + embeddings	≥ 0.85
Context recall	Намерено ли е всичко нужно за еталонния отговор?	въпрос, контекст, еталон	≥ 0.80
Context precision	Подредени ли са полезните парчета най-отгоре?	въпрос, контекст, еталон	≥ 0.80

📊

Как се чете таблото (примерни, симулирани стойности)

Faithfulness 0.94 ✅ · Answer relevancy 0.91 ✅ · Context recall 0.78 ⚠️ · Context precision 0.88 ✅. Ниският recall след добавяне на нови документи обикновено значи, че те са нарязани или индексирани различно. Провери парчетата им и преиндексирай — по-голямо top_k само маскира проблема. Праговете са отправна точка, не стандарт: определи своите от първите замервания.

Оценка с RAGAS 0.4 и локален съдия

Ollama има OpenAI-съвместим адрес, затова съдията се свързва с обикновения OpenAI клиент — но заявките не излизат от твоята машина. Защо по-силен съдия? Слабият модел оценява непоследователно и понякога не спазва формата, който RAGAS очаква.

Python · ragas_evaluation.py

import asyncio, csv, sys
from datetime import datetime
from pathlib import Path

from openai import AsyncOpenAI
from ragas.llms import llm_factory
from ragas.embeddings import OpenAIEmbeddings
from ragas.metrics.collections import (Faithfulness, AnswerRelevancy,
                                       ContextRecall, ContextPrecision)
from rag_core import advanced_rag_retrieve, advanced_rag_answer   # от урок 01-04a

# ── Съдия: локален модел през OpenAI-съвместимия адрес на Ollama ──
client = AsyncOpenAI(base_url="http://localhost:11434/v1", api_key="ollama")  # клиентът иска ключ, Ollama не го проверява
judge = llm_factory("qwen3:14b", client=client)            # по-силен от модела, който отговаря
embeddings = OpenAIEmbeddings(client=client, model="nomic-embed-text")

faithfulness = Faithfulness(llm=judge)
relevancy = AnswerRelevancy(llm=judge, embeddings=embeddings)
recall = ContextRecall(llm=judge)
precision = ContextPrecision(llm=judge)

THRESHOLDS = {"faithfulness": 0.90, "answer_relevancy": 0.85,
              "context_recall": 0.80, "context_precision": 0.80}

# ── Златен набор: въпрос + еталонен отговор. Разширявай с реални въпроси. ──
EVAL_SET = [
    {"question": "Кой плаща първите два работни дни от болничния?",
     "reference": "Работодателят, по 70% от среднодневното брутно възнаграждение (чл. 40, ал. 5 КСО)."},
    {"question": "Колко е обезщетението от НОИ при общо заболяване?",
     "reference": "80% от среднодневното брутно възнаграждение за 18-те календарни месеца преди "
                  "болничния, но не повече от среднодневното нетно (чл. 41 КСО)."},
    {"question": "Какво урежда чл. 87 от ЗЗД?",
     "reference": "Развалянето на двустранен договор, когато длъжникът не изпълни по причина, за която отговаря."},
    {"question": "Какви вреди се обезщетяват по чл. 82 ЗЗД?",
     "reference": "Претърпяната загуба и пропуснатата полза, ако са пряка и непосредствена последица от "
                  "неизпълнението и са могли да се предвидят; при недобросъвестност — всички преки вреди."},
    {"question": "С каква ставка на ДДС е износът на стоки извън ЕС?",
     "reference": "С нулева ставка (чл. 28 ЗДДС)."},
]

async def score_one(item: dict) -> dict:
    q, ref = item["question"], item["reference"]
    docs = await asyncio.to_thread(advanced_rag_retrieve, q, top_k=5)
    ctx = [d["text"] for d in docs]
    answer = (await asyncio.to_thread(advanced_rag_answer, q))["answer"]
    r = {
        "faithfulness": await faithfulness.ascore(user_input=q, response=answer, retrieved_contexts=ctx),
        "answer_relevancy": await relevancy.ascore(user_input=q, response=answer),
        "context_recall": await recall.ascore(user_input=q, retrieved_contexts=ctx, reference=ref),
        "context_precision": await precision.ascore(user_input=q, reference=ref, retrieved_contexts=ctx),
    }
    return {"question": q, **{k: float(v.value) for k, v in r.items()}}

async def run_eval(out_dir: str = "./ragas_results") -> dict:
    rows = [await score_one(item) for item in EVAL_SET]    # последователно — пести паметта на GPU
    scores = {m: round(sum(r[m] for r in rows) / len(rows), 3) for m in THRESHOLDS}
    failures = {m: s for m, s in scores.items() if s < THRESHOLDS[m]}

    Path(out_dir).mkdir(parents=True, exist_ok=True)       # CSV за тренда във времето
    stamp = datetime.now().strftime("%Y%m%d_%H%M")
    with open(f"{out_dir}/ragas_{stamp}.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=rows[0].keys())
        w.writeheader(); w.writerows(rows)

    for m, s in scores.items():
        print(f"{'✅' if s >= THRESHOLDS[m] else '❌'} {m}: {s:.3f} (праг ≥ {THRESHOLDS[m]})")
    return {"scores": scores, "failures": failures, "timestamp": stamp}

if __name__ == "__main__":
    result = asyncio.run(run_eval())
    sys.exit(1 if result["failures"] else 0)               # код 1 → сигнал за cron / CI

⚠️

Стар код в интернет

Много примери ползват from ragas.metrics import faithfulness, LangchainLLMWrapper и evaluate(Dataset…) с колони question / answer / contexts / ground_truth. В RAGAS 0.4 това още работи, но е остаряло и ще отпадне във v1.0. Новите полета са user_input, response, retrieved_contexts, reference.

⛔

Еталонът се проверява от човек

Златният набор е мярката за всичко останало. Грешен еталон („болничният е до 6 месеца“, „членове 45–46“ — такива имаше в стария урок) прави метриките безсмислени. Всеки еталонен отговор — с източник и проверен от специалист.

Нощна оценка със сигнал

Оценката има смисъл, ако върви сама и редовно — особено след обновяване на документите. Скриптът пуска оценката, а при код 1 праща сигнал в чата на екипа.
bash · ragas_monitor.sh
```
#!/usr/bin/env bash
# Всяка нощ в 03:00 (crontab -e):
# 0 3 * * * /path/to/rag/ragas_monitor.sh >> /path/to/rag/logs/ragas.log 2>&1
set -u
cd /path/to/rag && source .venv/bin/activate
source ./.env                       # cron не вижда твоите променливи: ALERT_WEBHOOK_URL=...

if python3 ragas_evaluation.py; then
  echo "$(date -Is) RAGAS OK"
else
  curl -s -X POST "$ALERT_WEBHOOK_URL" -H 'Content-Type: application/json' \
       -d '{"text":"⚠️ RAG: метрика под прага. Виж ragas_results/"}'
  echo "$(date -Is) RAGAS ALERT"
fi
```
🖥️
Локално — без облак и без такси на заявка
С локален съдия и локални embeddings данните не напускат машината, а оценката не струва нищо на заявка — само ток и време. Колко време отнема зависи от модела и хардуера: замери първото пускане и настрой часа на задачата така, че да не се бие с работното натоварване.