用 RAGFlow 检索接口定位知识库回答问题

RAGFlow 是开源的 RAG（检索增强生成）引擎，擅长解析 PDF、表格等复杂文档。 回答不好时，先要分清是“没检索到”还是“检索到了但模型没用好”。这篇做 rf_retrieve.py：只调用检索接口，列出命中的片段、综合相似度、关键词相似度、向量相似度和来源文件，方便调整阈值和权重。

先看清成品

• 对 demo.ragflow.io 真实调用：无效 Key 返回 HTTP 401，body 的 code 也是 401，message 是一段 Python 异常的字符串。
• 脚本同时检查 HTTP 状态和 body code，报错时附带“后端是 1.0 Go 还是之前的 Python”，避免对着错误版本的文档排查。
• 6 项 pytest 通过（含 HTTP 200 但 code=102 的业务错误）。

合规提醒：RAGFlow 是 Apache 2.0 开源项目。本文只用官方 HTTP API 和你自己实例的 API Key（在 RAGFlow 网页的 API 设置中创建）。Key 只放服务端。demo.ragflow.io 是官方的公开演示站，我只用无效 Key 对它发了几个请求来确认错误格式，没有注册、上传或压测；不要把它当成自己的生产服务，也不要上传敏感文档。

运行说明：RAGFlow 官方推荐至少 4 核 CPU、16 GB 内存、50 GB 磁盘，并用 Docker 部署。box 没有 Docker 权限，所以没有部署 RAGFlow。真实跑过的是：用无效 Key 调用 demo.ragflow.io（2026-10-21 21:31–21:33，UTC+8，返回原样贴在文中），以及全部本地代码。正常返回来自我写的 mock，字段照官方 HTTP API 参考。版本提醒：2026-10-16 发布了 v1.0.0-rc2，main 分支的 API 文档已经按 1.0（Go 后端）写，有几个旧接口被移除；demo 站的响应没有 1.0 才有的 X-API-Source: go 头。先确认你的版本，再看对应版本的文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

84-ragflow-A-src/
├── rf_retrieve.py
└── test_rf_retrieve.py

第 1 步：脚本

"""调用 RAGFlow 的检索接口 POST /api/v1/retrieval，只看“检索到了什么”，不经过大模型。
用来调试知识库：问题命中了哪些片段、相似度多少、来自哪个文件。

RAGFlow 的响应同时有 HTTP 状态码和 body 里的业务 code（0 才是成功），官方要求两者分开检查。"""
import argparse, json, os, sys
import requests

UA = "binbho-tutorial/1.0"
BODY_CODES = {101: "参数错误", 102: "数据无效或缺失", 108: "权限错误", 109: "认证错误", 401: "未授权（API Key 无效）"}


class RagflowError(Exception):
    pass


def backend_of(resp):
    """RAGFlow 1.0 的 Go 后端会带 X-API-Source: go；没有这个头多半是 1.0 之前的 Python 后端。"""
    return "go（1.0）" if (getattr(resp, "headers", {}) or {}).get("X-API-Source") == "go" else "python（1.0 之前）"


def post(base, key, path, payload, session=None):
    s = session or requests
    r = s.post(base.rstrip("/") + path, json=payload, timeout=60,
               headers={"Authorization": f"Bearer {key.strip()}", "User-Agent": UA})
    try:
        d = r.json()
    except ValueError:
        raise RagflowError(f"HTTP {r.status_code}，不是 JSON：{r.text[:120]}")
    if r.status_code != 200 or d.get("code") != 0:
        code = d.get("code")
        raise RagflowError(f"HTTP {r.status_code} code={code}（{BODY_CODES.get(code, '未知')}）：{d.get('message')}"
                           f"  [后端：{backend_of(r)}]")
    return d["data"]


def retrieve(base, key, question, dataset_ids, threshold=0.2, kw_weight=0.7, page_size=5, session=None):
    return post(base, key, "/api/v1/retrieval", {
        "question": question, "dataset_ids": dataset_ids, "page": 1, "page_size": page_size,
        "similarity_threshold": threshold, "keywords_similarity_weight": kw_weight, "highlight": True}, session)


def report(data):
    lines = [f"命中 {data.get('total', 0)} 条，显示 {len(data.get('chunks', []))} 条"]
    for i, c in enumerate(data.get("chunks", []), 1):
        lines.append(f"[{i}] {c.get('document_keyword')}  综合 {c['similarity']:.3f}"
                     f"（关键词 {c.get('term_similarity', 0):.3f} / 向量 {c.get('vector_similarity', 0):.3f}）")
        lines.append("    " + c["content"].replace("\n", " ")[:100])
    for d in data.get("doc_aggs", []):
        lines.append(f"  文件 {d['doc_name']}：{d['count']} 条")
    return "\n".join(lines)


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("question"); ap.add_argument("--dataset", action="append", required=True)
    ap.add_argument("--threshold", type=float, default=0.2); ap.add_argument("--kw-weight", type=float, default=0.7)
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args(argv)
    base, key = os.environ.get("RAGFLOW_BASE_URL", "http://localhost"), os.environ.get("RAGFLOW_API_KEY", "")
    if not key.strip():
        print("缺少 RAGFLOW_API_KEY", file=sys.stderr); return 1
    try:
        data = retrieve(base, key, a.question, a.dataset, a.threshold, a.kw_weight)
    except (RagflowError, requests.RequestException) as e:
        print(f"检索失败：{e}", file=sys.stderr); return 1
    print(json.dumps(data, ensure_ascii=False, indent=2) if a.json else report(data))
    return 0


if __name__ == "__main__":
    sys.exit(main())

第 2 步：测试（mock）

"""mock 是我写的，结构照 RAGFlow 官方 HTTP API 参考里 Retrieve chunks 的示例。"""
import pytest
import rf_retrieve as r


class R:
    def __init__(self, code, body):
        self.status_code, self._b, self.text, self.headers = code, body, str(body), {}
    def json(self):
        if isinstance(self._b, str):
            raise ValueError
        return self._b


class S:
    def __init__(self, resp):
        self.resp, self.sent = resp, None
    def post(self, url, json=None, timeout=None, headers=None):
        self.sent = (url, json, headers); return self.resp


OK = {"code": 0, "data": {"chunks": [{"content": "入职满一年后，每年享有 10 天带薪年假。", "document_keyword": "员工手册.md",
      "id": "c1", "similarity": 0.8123, "term_similarity": 0.9, "vector_similarity": 0.61}],
      "doc_aggs": [{"count": 1, "doc_id": "d1", "doc_name": "员工手册.md"}], "total": 1}}


def test_payload_and_report():
    s = S(R(200, OK))
    data = r.retrieve("http://localhost/", " k\n", "年假几天", ["ds1"], session=s)
    url, body, h = s.sent
    assert url == "http://localhost/api/v1/retrieval" and body["dataset_ids"] == ["ds1"]
    assert body["keywords_similarity_weight"] == 0.7 and "vector_similarity_weight" not in body
    assert h["Authorization"] == "Bearer k"
    out = r.report(data)
    assert "员工手册.md  综合 0.812（关键词 0.900 / 向量 0.610）" in out and "文件 员工手册.md：1 条" in out


def test_http_200_but_business_error():
    with pytest.raises(r.RagflowError, match="code=102"):
        r.retrieve("http://h", "k", "q", [], session=S(R(200, {"code": 102, "message": "`dataset_ids` is required."})))


def test_real_401_shape():   # 2026-10-21 对 demo.ragflow.io 实测
    body = {"code": 401, "data": None, "message": "<Unauthorized '401: Unauthorized'>"}
    with pytest.raises(r.RagflowError, match="API Key 无效"):
        r.retrieve("http://h", "bad", "q", ["x"], session=S(R(401, body)))


def test_non_json():
    with pytest.raises(r.RagflowError, match="不是 JSON"):
        r.retrieve("http://h", "k", "q", ["x"], session=S(R(502, "<html>")))


def test_empty_result():
    assert r.report({"chunks": [], "doc_aggs": [], "total": 0}).startswith("命中 0 条")


def test_backend_detection():
    resp = R(401, {}); assert r.backend_of(resp).startswith("python")
    resp.headers = {"X-API-Source": "go"}; assert r.backend_of(resp).startswith("go")

第 3 步：运行

# 真实请求（2026-10-21 21:31 UTC+8）
$ curl -s -D - https://demo.ragflow.io/api/v1/datasets -H 'Authorization: Bearer ragflow-invalid-demo'
server: openresty/1.31.1.1
content-type: application/json
{"code":401,"data":null,"message":"<Unauthorized '401: Unauthorized'>"}   ← HTTP 401（不带 Key 结果相同；响应头里没有 X-API-Source）

$ RAGFLOW_BASE_URL=https://demo.ragflow.io RAGFLOW_API_KEY=ragflow-invalid-demo python rf_retrieve.py "年假有几天" --dataset 00000000000000000000000000000000
检索失败：HTTP 401 code=401（未授权（API Key 无效））：<Unauthorized '401: Unauthorized'>  [后端：python（1.0 之前）]
$ python -m pytest -q
6 passed

# 有自己实例时的输出格式（来自 mock）：
命中 1 条，显示 1 条
[1] 员工手册.md  综合 0.812（关键词 0.900 / 向量 0.610）
    入职满一年后，每年享有 10 天带薪年假。
  文件 员工手册.md：1 条

怎么检查结果？

• 同一个问题换几种问法（口语、书面、带错别字）各跑一次，看命中是否稳定。
• 关键词相似度高、向量相似度低：多半是字面重合但意思不对；反过来是同义改写。用 --kw-weight 调整两者比例。
• --threshold 调高会减少噪音，也可能把正确片段过滤掉。用 C 篇或 83C 那样的固定问题集来定值。

调试示例

先确认版本，再看文档。 我一开始照着 GitHub main 分支的 API 参考写代码，后来注意到两件事：

1. 最新发布是 v1.0.0-rc2（2026-10-16），main 分支文档写的是 1.0：公开 API 改由 Go 实现，POST /api/v1/chats_openai/{chat_id}/chat/completions 和 POST /api/v1/chats/{chat_id}/completions 这类旧路径在 1.0 中被移除；文档还提醒“部分校验的业务码、错误信息和认证的 HTTP 状态和 Python 版不同”。
2. demo.ragflow.io 的响应头只有 server: openresty/… 和 content-type，没有 X-API-Source: go，说明它跑的不是 1.0 的 Go 后端。

如果拿 1.0 的文档去调 1.0 之前的实例（或反过来），会遇到 404 或字段不认识的问题，很容易误以为是 Key 或网络的问题。所以我加了 backend_of()：出错时在信息末尾标出后端类型。这个判断依据是官方说 Go 响应会带这个头；如果你前面的反向代理删掉了它，判断会不准。

另外，demo 站返回的 message 是 <Unauthorized '401: Unauthorized'>，一看就是后端异常对象转成的字符串。官方也建议按 code 分支，不要匹配 message 文本。

常见问题

模板规则

• 先单独测检索，再测问答。
• HTTP 状态和 body code 都要检查。
• 确认后端版本再看文档。
• 阈值和权重用问题集来定。

项目资源

检索调对了，问答就成功了一大半。