RAG：让本地模型读取 AlgoLab 知识RAG: Retrieval-Augmented Generation for a Local LLM

import json
import urllib.request

API = "http://127.0.0.1:4000/api"


def post(path, payload):
    request = urllib.request.Request(
        API + path,
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=120) as response:
        return json.load(response)


def main():
    result = post("/rag/chunk", {
        "modules_from": 8,
        "modules_to": 8,
        "chunk_size": 520,
        "overlap": 80,
        "sample_limit": 12,
    })
    print("=== Real No.8 chunking ===")
    for key in ("module_range", "document_count", "chunk_count",
                "source_chars", "stored_chars", "duplicated_chars"):
        print(f"{key:18s}: {result.get(key)}")
    print("execution         :", result.get("execution"))

    samples = result.get("sample_chunks", [])
    print(f"returned samples  : {len(samples)}")
    for chunk in samples:
        start = chunk.get("start", chunk.get("char_start"))
        end = chunk.get("end", chunk.get("char_end"))
        digest = chunk.get("content_hash", chunk.get("source_hash"))
        print(f"\n{chunk.get('chunk_id')} offset={start}:{end} hash={digest}")
        print(chunk.get("text", ""))

    assert result.get("ok") is True
    assert result.get("chunk_count", 0) > 0
    assert result.get("execution", {}).get("location") == "algolab_backend"
    assert "index_id" not in result, "chunk experiment must not build an index"
    assert samples, "Backend must return observable real chunks"
    print("\nchecks passed: No.8 was really chunked; no model or index was used")


if __name__ == "__main__":
    main()

import json
import math
import urllib.request

API = "http://127.0.0.1:4000/api"
QUERY = "为什么 Tool Calling 不能只看 loss？"


def post(path, payload):
    request = urllib.request.Request(
        API + path,
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=180) as response:
        return json.load(response)


def main():
    index = post("/rag/index", {
        "modules_from": 1, "modules_to": 13,
        "chunk_size": 520, "overlap": 80,
    })
    result = post("/rag/retrieve", {
        "index_id": index["index_id"],
        "query": QUERY,
        "top_k": 4,
        "context_budget": 2200,
    })

    preview = result.get("query_vector_preview", [])
    print("=== Real query embedding and Top-K ===")
    print("index             :", index.get("index_id"))
    print("embedding backend :", result.get("embedding_backend"))
    print("dimension         :", result.get("dimension", index.get("dimension")))
    print("query norm        :", result.get("query_norm"))
    print("vector preview    :", preview)
    print("context usage     :", result.get("context_used"), "/", result.get("context_budget"))
    print("execution         :", result.get("execution"))

    for hit in result.get("hits", []):
        print(
            f"{hit.get('rank')}. No.{hit.get('module_no')} "
            f"cos={hit.get('score')} {hit.get('section_title')}"
        )
    print("\nBudget trace:")
    for step in result.get("budget_trace", []):
        print(step)
    print("\nContext actually assembled:\n", result.get("context", ""))

    assert result.get("ok") is True
    assert result.get("index_id") == index.get("index_id")
    assert result.get("execution", {}).get("location") == "algolab_backend"
    if preview and result.get("query_norm") is not None:
        assert math.isfinite(float(result["query_norm"]))
    assert result.get("hits"), "real retrieval must return ranked hits"
    print("\nchecks passed: the result came from the active real index")


if __name__ == "__main__":
    main()

import argparse
import json
import os
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

BACKEND = os.getenv(
    "ALGOLAB_API_URL", "https://algolab.coding-x.tech/api/rag"
).rstrip("/")
QUERY = "为什么 Tool Calling 不能只看 loss？"


def request_json(base_url, path, payload=None, timeout=600):
    data = None if payload is None else json.dumps(
        payload, ensure_ascii=False
    ).encode("utf-8")
    request = Request(
        base_url + path,
        data=data,
        headers={"Content-Type": "application/json; charset=utf-8"},
        method="GET" if payload is None else "POST",
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            return json.load(response)
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"{base_url}{path} HTTP {exc.code}: {detail}") from exc
    except URLError as exc:
        raise RuntimeError(f"无法连接 {base_url}{path}: {exc.reason}") from exc


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--local-runner-url", default="http://127.0.0.1:4890")
    parser.add_argument("--model-path", default="")
    parser.add_argument("--adapter-path", default="")
    parser.add_argument("--embedding-model-path", default="")
    args = parser.parse_args()
    runner = args.local_runner_url.rstrip("/")

    backend_status = request_json(BACKEND, "/status")
    try:
        runner_status = request_json(runner, "/status", timeout=10)
    except RuntimeError as error:
        runner_status = {"ok": False, "message": str(error)}
    print("=== 0. Backend + browser/direct Local Runner contract ===")
    print(json.dumps(
        {"backend": backend_status, "local_runner": runner_status},
        ensure_ascii=False, indent=2,
    ))

    use_local_embedding = bool(args.embedding_model_path.strip())
    index_payload = {
        "modules_from": 1,
        "modules_to": 13,
        "chunk_size": 520,
        "overlap": 80,
    }
    if use_local_embedding:
        index_payload["embedding_model_path"] = args.embedding_model_path
        index = request_json(runner, "/rag/index", index_payload)
    else:
        index = request_json(BACKEND, "/index", index_payload)
    print("\n=== 1. Real index ===")
    print(json.dumps(index, ensure_ascii=False, indent=2))
    assert index.get("ok") is True and index.get("index_id")

    retrieval_payload = {
        "query": QUERY,
        "top_k": 4,
        "context_budget": 2200,
    }
    if use_local_embedding:
        retrieval = request_json(runner, "/rag/retrieve", retrieval_payload)
    else:
        retrieval_payload["index_id"] = index["index_id"]
        retrieval = request_json(BACKEND, "/retrieve", retrieval_payload)
    print("\n=== 2. Real retrieval and context ===")
    print(json.dumps(retrieval, ensure_ascii=False, indent=2))
    assert retrieval.get("hits")
    assert retrieval.get("context"), "retrieval must expose the exact context"
    assert retrieval["context_used"] <= retrieval["context_budget"]

    if not args.model_path.strip():
        print("\n=== 3. Generation: NOT RUN ===")
        print("CPU retrieval 已完成；传入 --model-path 才直连本机 Qwen。")
        return

    answer = request_json(runner, "/rag/generate", {
        "coordinator": "standalone_browser_direct_contract",
        "query": QUERY,
        "context": retrieval["context"],
        "citations": retrieval["citations"],
        "index_id": retrieval.get("index_id", index.get("index_id")),
        "embedding_backend": retrieval.get(
            "embedding_backend", index.get("embedding_backend")
        ),
        "retrieval_ms": retrieval.get("retrieval_ms"),
        "context_used": retrieval.get("context_used"),
        "context_budget": retrieval.get("context_budget"),
        "context_remaining": retrieval.get("context_remaining"),
        "context_budget_unit": retrieval.get("context_budget_unit", "characters"),
        "budget_measurement": retrieval.get("budget_measurement", {}),
        "budget_trace": retrieval.get("budget_trace", []),
        "model_path": args.model_path,
        "adapter_path": args.adapter_path,
        "max_new_tokens": 420,
        "temperature": 0.2,
    }, timeout=1800)
    print("\n=== 3. Real local-model answer ===")
    print(answer.get("answer"))
    print("citations:", answer.get("citations"))
    print("valid references:", answer.get("valid_references"))
    assert answer.get("answer")
    assert answer.get("context") == retrieval.get("context")
    assert answer.get("context_sha256")


if __name__ == "__main__":
    main()

import json
import urllib.request

API = "http://127.0.0.1:4000/api"


def post(path, payload):
    request = urllib.request.Request(
        API + path,
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=900) as response:
        return json.load(response)


def main():
    index = post("/rag/index", {
        "modules_from": 1, "modules_to": 13,
        "chunk_size": 520, "overlap": 80,
    })
    result = post("/rag/evaluate", {
        "index_id": index["index_id"],
        "top_k": 4,
        "context_budget": 2200,
    })

    print("=== Real layered RAG evaluation ===")
    print("index:", index.get("index_id"))
    print("summary:", json.dumps(result.get("summary", {}), ensure_ascii=False, indent=2))
    for case in result.get("cases", []):
        print("\nquery:", case.get("query"))
        print("gold:", case.get("gold_sources"))
        print("retrieved:", case.get("retrieved_sources", case.get("hits")))
        print("metrics:", case.get("metrics"))
        print("generation:", case.get("generation_status", "not reported"))

    summary = result.get("summary", {})
    for name in ("recall_at_k", "mrr", "context_precision"):
        value = summary.get(name)
        assert value is not None and 0.0 <= float(value) <= 1.0
    assert result.get("generation_ran") is False
    assert summary.get("faithfulness") is None
    assert result.get("execution", {}).get("location") == "algolab_backend"
    assert result.get("cases"), "Backend must execute real evaluation cases"
    print("\nchecks passed: CPU metrics came from executed cases; generation was not run")


if __name__ == "__main__":
    main()