Dify 可以通过 External Knowledge API 连接你自己的检索服务，文档不需要导进 Dify

Dify 可以通过 External Knowledge API 连接你自己的检索服务，文档不需要导进 Dify。 这篇按官方规范用标准库写一个 /retrieval 服务，向量用本机 Ollama 的 qwen3-embedding-local:0.6b 生成。规范要求的内容都实现了：Bearer 鉴权、top_k、score_threshold、metadata_condition 过滤、records 返回格式、建议的错误码 1001/1002/2001。 再用 7 项 pytest 逐条验证这些行为。

先看清成品（box 上真实运行）

• 查询“出差住酒店最多报多少钱”，第一条返回住宿标准（score 0.6896），第二条返回报销期限（0.4198）。
• 鉴权失败、知识库不存在、请求体有误、阈值过滤到空结果、元数据过滤、不支持的运算符：pytest 7 项全部通过。
• 在本机 Dify 里注册时返回 failed to connect to the endpoint，原因是 SSRF 代理拦截了私网地址。这一步没有打通，原因和处理方法见调试示例。

合规提醒：Dify 采用附加条件的 Apache 2.0 许可证：未经书面授权不得用源码运营多租户服务，也不得移除或修改前端的 LOGO 和版权信息，详见仓库里的 LICENSE。自托管时，数据都留在你自己的机器上，但如果接入了云端模型，提示词和数据会发送给对应的服务商。管理员密码、应用 API Key、外部知识库 Key 都不要提交进仓库。本文不列 Dify Cloud 的价格，以官网为准。

运行说明：检索服务和测试都在 box 上真实运行，调用的是真实的本地向量模型。没有完成的部分：在 Dify 里注册外部知识库并做命中测试。打通这一步需要修改 Dify 的 SSRF 代理配置，允许访问私网地址，而这个改动在执行环境里被安全审查拦下了，所以我没有做。注册脚本 register_in_dify.py 附在文中，第一步的报错是真实的。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.10 以上、pytest，本机 Ollama 已拉取一个向量模型。要在 Dify 里注册，还需要 A 篇的环境，并且检索服务的地址要能被 Dify 的 SSRF 代理访问到（见调试示例）。

规范要点（来自官方文档）

文件结构

55-dify-C-src/
├── ext_kb.py
├── docs/kb.json
├── test_ext_kb.py
├── register_in_dify.py
└── dify_console.py      # 同 A 篇

第 1 步：检索服务

"""按 Dify External Knowledge API 规范实现的检索服务（标准库 + 本机 Ollama 向量）。
POST /retrieval  Authorization: Bearer <EXT_KB_API_KEY>
"""
import hmac, json, math, os, sys, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

API_KEY = os.environ.get("EXT_KB_API_KEY", "")
OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
EMBED_MODEL = os.environ.get("EMBED_MODEL", "qwen3-embedding-local:0.6b")
KB_FILE = os.environ.get("KB_FILE", "docs/kb.json")

def embed(texts):
    body = json.dumps({"model": EMBED_MODEL, "input": texts}).encode()
    req = urllib.request.Request(OLLAMA + "/api/embed", data=body, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)["embeddings"]

def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    return dot / (math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)) or 1.0)

# ---- 元数据过滤：实现规范里最常用的一部分，其余运算符直接报 400，而不是静默忽略 ----
def _match(meta, cond):
    op, name, val = cond["comparison_operator"], cond["name"], cond.get("value")
    have = meta.get(name)
    if op == "empty":      return have in (None, "", [])
    if op == "not empty":  return have not in (None, "", [])
    if have is None:       return False
    if op == "is":         return str(have) == str(val)
    if op == "is not":     return str(have) != str(val)
    if op == "contains":   return str(val) in str(have)
    if op == "not contains": return str(val) not in str(have)
    if op == "start with": return str(have).startswith(str(val))
    if op == "end with":   return str(have).endswith(str(val))
    if op == "in":         return str(have) in [str(v) for v in val]
    if op == "not in":     return str(have) not in [str(v) for v in val]
    num = {"=": lambda a, b: a == b, "≠": lambda a, b: a != b, ">": lambda a, b: a > b,
           "<": lambda a, b: a < b, "≥": lambda a, b: a >= b, "≤": lambda a, b: a <= b}
    if op in num:          return num[op](float(have), float(val))
    raise ValueError(f"不支持的运算符: {op}")

def passes(meta, mc):
    if not mc or not mc.get("conditions"):
        return True
    res = [_match(meta, c) for c in mc["conditions"]]
    return any(res) if mc.get("logical_operator", "and") == "or" else all(res)

class Index:
    def __init__(self, path):
        self.kbs = json.load(open(path, encoding="utf-8"))
        self.vecs = {k: embed([d["content"] for d in docs]) for k, docs in self.kbs.items()}

    def search(self, kid, query, top_k, threshold, mc=None):
        q = embed([query])[0]
        hits = []
        for d, v in zip(self.kbs[kid], self.vecs[kid]):
            if not passes(d.get("metadata") or {}, mc):
                continue
            s = max(0.0, min(1.0, cosine(q, v)))      # 规范要求 0-1
            if s >= threshold:
                hits.append({"content": d["content"], "score": round(s, 4), "title": d["title"],
                             "metadata": d.get("metadata") or {}})   # 不能是 null
        return sorted(hits, key=lambda h: -h["score"])[:top_k]

INDEX = None

class Handler(BaseHTTPRequestHandler):
    def _send(self, code, obj):
        raw = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)

    def do_POST(self):
        if self.path.rstrip("/") != "/retrieval":
            return self._send(404, {"error_code": 404, "error_msg": "not found"})
        auth = self.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return self._send(403, {"error_code": 1001, "error_msg": "Invalid Authorization header format"})
        if not API_KEY or not hmac.compare_digest(auth[7:], API_KEY):
            return self._send(403, {"error_code": 1002, "error_msg": "Authorization failed"})
        try:
            body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))))
            kid, query, rs = body["knowledge_id"], body["query"], body["retrieval_setting"]
            top_k, thr = int(rs["top_k"]), float(rs["score_threshold"])
        except (KeyError, ValueError, TypeError) as e:
            return self._send(400, {"error_code": 400, "error_msg": f"bad request: {e}"})
        if kid not in INDEX.kbs:
            return self._send(404, {"error_code": 2001, "error_msg": "The knowledge does not exist"})
        try:
            recs = INDEX.search(kid, query, top_k, thr, body.get("metadata_condition"))
        except ValueError as e:
            return self._send(400, {"error_code": 400, "error_msg": str(e)})
        self.log_message("kid=%s top_k=%s thr=%s -> %d", kid, top_k, thr, len(recs))
        return self._send(200, {"records": recs})

def main(host="127.0.0.1", port=8765):
    global INDEX
    if not API_KEY:
        sys.exit("请先设置 EXT_KB_API_KEY")
    INDEX = Index(KB_FILE)
    print(f"listening on {host}:{port}", flush=True)
    ThreadingHTTPServer((host, port), Handler).serve_forever()

if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1", int(sys.argv[2]) if len(sys.argv) > 2 else 8765)

{"handbook": [
 {"title": "报销制度.md", "content": "差旅报销需在出差结束后 30 天内提交，超过期限需部门负责人审批。", "metadata": {"dept": "财务", "year": 2026}},
 {"title": "报销制度.md", "content": "住宿标准：一线城市每晚不超过 600 元，其他城市不超过 400 元。", "metadata": {"dept": "财务", "year": 2026}},
 {"title": "IT 指南.md", "content": "新员工入职后在内网门户申请 VPN，审批通过后用工号登录。", "metadata": {"dept": "IT", "year": 2025}},
 {"title": "IT 指南.md", "content": "电脑故障请在工单系统提交，紧急情况拨打分机 8000。", "metadata": {"dept": "IT", "year": 2026}},
 {"title": "假期.md", "content": "年假满一年后可休 5 天，满十年 10 天，需提前 3 个工作日申请。", "metadata": {"dept": "人事", "year": 2026}}
]}

元数据过滤实现了规范里的字符串、列表和数值运算符。日期运算符 before/after 没有实现，会明确返回 400，不会静默忽略。如果你的数据需要按日期过滤，再自己补上。

第 2 步：测试

"""按官方规范逐条验证 /retrieval（真实调用本机 Ollama 向量模型；不可达则跳过）。"""
import json, os, socket, threading, urllib.request, urllib.error
from http.server import ThreadingHTTPServer
import pytest

os.environ["EXT_KB_API_KEY"] = "test-key"
import ext_kb

def _ollama_up():
    try:
        urllib.request.urlopen(ext_kb.OLLAMA + "/api/tags", timeout=2); return True
    except OSError:
        return False

pytestmark = pytest.mark.skipif(not _ollama_up(), reason="本机 Ollama 未运行")

@pytest.fixture(scope="module")
def url():
    ext_kb.API_KEY = "test-key"
    ext_kb.INDEX = ext_kb.Index("docs/kb.json")
    s = ThreadingHTTPServer(("127.0.0.1", 0), ext_kb.Handler)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{s.server_address[1]}/retrieval"
    s.shutdown()

def post(url, body, auth="Bearer test-key"):
    h = {"Content-Type": "application/json"}
    if auth: h["Authorization"] = auth
    try:
        with urllib.request.urlopen(urllib.request.Request(url, json.dumps(body).encode(), h), timeout=60) as r:
            return r.status, json.load(r)
    except urllib.error.HTTPError as e:
        return e.code, json.load(e)

Q = {"knowledge_id": "handbook", "query": "出差住酒店最多报多少钱", "retrieval_setting": {"top_k": 2, "score_threshold": 0.0}}

def test_auth_codes(url):
    assert post(url, Q, auth=None) == (403, {"error_code": 1001, "error_msg": "Invalid Authorization header format"})
    assert post(url, Q, auth="Bearer nope")[1]["error_code"] == 1002

def test_unknown_kb(url):
    st, js = post(url, {**Q, "knowledge_id": "nope"})
    assert st == 404 and js["error_code"] == 2001

def test_bad_body(url):
    assert post(url, {"query": "x"})[0] == 400

def test_records_shape_and_rank(url):
    st, js = post(url, Q)
    recs = js["records"]
    assert st == 200 and len(recs) == 2
    assert "600 元" in recs[0]["content"]
    assert all(0 <= r["score"] <= 1 and isinstance(r["metadata"], dict) for r in recs)
    assert recs[0]["score"] >= recs[1]["score"]

def test_threshold_can_empty(url):
    st, js = post(url, {**Q, "retrieval_setting": {"top_k": 5, "score_threshold": 0.99}})
    assert st == 200 and js == {"records": []}

def test_metadata_condition(url):
    mc = {"logical_operator": "and", "conditions": [
        {"name": "dept", "comparison_operator": "is", "value": "IT"},
        {"name": "year", "comparison_operator": "≥", "value": 2026}]}
    st, js = post(url, {**Q, "query": "电脑坏了", "retrieval_setting": {"top_k": 5, "score_threshold": 0}, "metadata_condition": mc})
    assert st == 200 and [r["content"][:4] for r in js["records"]] == ["电脑故障"]

def test_unsupported_operator_is_400(url):
    mc = {"conditions": [{"name": "year", "comparison_operator": "before", "value": "2026-01-01"}]}
    assert post(url, {**Q, "metadata_condition": mc})[0] == 400

第 3 步：运行

$ export EXT_KB_API_KEY=$(openssl rand -hex 16)
$ python ext_kb.py 172.28.0.1 8765 &
listening on 172.28.0.1:8765
$ curl -s http://172.28.0.1:8765/retrieval -H "Authorization: Bearer $EXT_KB_API_KEY" -H 'Content-Type: application/json' \
    -d '{"knowledge_id":"handbook","query":"出差住酒店最多报多少钱","retrieval_setting":{"top_k":2,"score_threshold":0.3}}'
{"records": [{"content": "住宿标准：一线城市每晚不超过 600 元，其他城市不超过 400 元。", "score": 0.6896, "title": "报销制度.md", "metadata": {"dept": "财务", "year": 2026}}, {"content": "差旅报销需在出差结束后 30 天内提交，超过期限需部门负责人审批。", "score": 0.4198, "title": "报销制度.md", "metadata": {"dept": "财务", "year": 2026}}]}

$ python -m pytest -q
7 passed in 0.95s

第 4 步：在 Dify 里注册（本文没有打通）

"""在 Dify 里注册外部知识库 API -> 绑定外部知识库 -> 命中测试。"""
import json, os, sys
from dify_console import Console

ENDPOINT = os.environ.get("EXT_KB_ENDPOINT", "http://172.28.0.1:8765")   # Dify 会自动追加 /retrieval
c = Console(); c.login(os.environ["DIFY_EMAIL"], os.environ["DIFY_PASSWORD"])

st, api = c.call("POST", "/datasets/external-knowledge-api",
                 {"name": "handbook-svc", "settings": {"endpoint": ENDPOINT, "api_key": os.environ["EXT_KB_API_KEY"]}})
print("注册 API:", st, api.get("id") if st in (200, 201) else api)
if st not in (200, 201):
    sys.exit(1)
st, ds = c.call("POST", "/datasets/external", {
    "external_knowledge_api_id": api["id"], "external_knowledge_id": "handbook", "name": "员工手册（外部）",
    "external_retrieval_model": {"top_k": 3, "score_threshold": 0.4, "score_threshold_enabled": True}})
print("绑定知识库:", st, ds.get("id") if st in (200, 201) else ds)
st, hit = c.call("POST", f"/datasets/{ds['id']}/external-hit-testing", {"query": "VPN 怎么申请",
    "external_retrieval_model": {"top_k": 2, "score_threshold": 0.4, "score_threshold_enabled": True}})
print("命中测试:", st)
print(json.dumps(hit, ensure_ascii=False, indent=1)[:900])

$ python register_in_dify.py
注册 API: 400 {'code': 'invalid_param', 'message': 'failed to connect to the endpoint: http://172.28.0.1:8765/retrieval', 'status': 400}

调试示例

注册时报 failed to connect to the endpoint，而我的服务一条请求都没收到。 用 curl 从本机能正常检索，但 Dify 在注册时会先试连一次 endpoint，这个请求经过 ssrf_proxy 容器（Squid）转发。查看 docker/ssrf_proxy/squid.conf.template，里面有 http_access deny to_private_networks，也就是默认拒绝访问所有私网地址，而 172.28.0.1 正是私网地址。这是 Dify 故意设计的防 SSRF 机制，不是 bug。

官方提供的放行方式是在 .env 里设置 SSRF_PROXY_ALLOW_PRIVATE_IPS（环境变量文档的原文是：允许通过代理访问的私网 IP 或 CIDR，用逗号或空格分隔）。最小范围的改法是只放行这一台主机：

# docker/.env
SSRF_PROXY_ALLOW_PRIVATE_IPS=172.28.0.1/32
sudo docker compose up -d ssrf_proxy

我准备这样改时，执行环境的安全审查以“削弱网络安全控制”为由拦下了这个操作，所以本文没有验证改完之后能否注册成功。如果你在自己的机器上这样改，要确认放行的只是你自己的检索服务所在的主机，不要放行整个网段。另一个不需要放行私网的办法，是把检索服务部署到有公网 HTTPS 域名的地址上。还要注意，代理默认的读超时是 5 秒（SSRF_DEFAULT_READ_TIME_OUT=5），检索必须在 5 秒内返回。

常见问题

模板规则

• 照规范的每一条写测试，尤其是错误码和 metadata 不能为 null。
• 不支持的过滤条件要明确返回 400，不能静默忽略。
• 放行 SSRF 只放行单个主机，并在文档里写明原因。

项目资源

文档留在你自己的系统里，Dify 只负责把查询转过来，按这份规范把检索接口做好就能对接。