(dify)如何使用dify自定义知识库【dify外部链接知识库】

蒲善思 · 2025-6-2 23:51:16

尝试dify自定义知识库

根据官网教程，可以从知识库的右上角外部知识库进行添加外部知识库
前往 “知识库” 页，点击右上角的 “外部知识库 API”，轻点 “添加外部知识库 API”。
按照页面提示，依次填写以下内容：

知识库的名称，允许自定义名称，用于区分所连接的不同外部知识 API；
API 接口地址，外部知识库的连接地址，示例 api-endpoint/retrieval；详细说明请参考外部知识库 API；
API Key，外部知识库连接密钥，详细说明请参考外部知识库 API；

因为APIEndpoint需要网络url地址，这里使用本地当作服务器进行尝试
1 使用python+flask框架构建本地后端

教程：python flask框架详解
1.1简单上手

from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello World'
if __name__ == '__main__':
app.run()

复制代码

在简单上手中，我们使用到了装饰器：@app.route('/')，要先了解装饰器，然后了解falsk这个装饰器以及其他类似的装饰器的用法。
1.2falsk的其他装饰器以及用法：

【扩展阅读，可跳过】
@app.route()

作用：将视图函数与指定的 URL 路径进行绑定。
示例：

@app.route('/') # 路由装饰器，绑定URL路径
def home():
return 'Hello, World!'

复制代码

@app.before_request()

作用：注册一个函数，在每个请求执行之前调用。适用于一些请求前的预处理，比如认证检查、日志记录等。
示例：

@app.before_request
def before_request():
print("This runs before every request.")

复制代码

@app.after_request()

作用：注册一个函数，在每个请求执行之后调用。适用于请求处理后的操作，如修改响应数据、日志记录等。
示例：

@app.after_request
def after_request(response):
print("This runs after each request.")
return response # 必须返回响应对象

复制代码

@app.errorhandler()

作用：注册一个函数，用于处理指定 HTTP 错误码的错误。例如，处理 404 页面未找到或 500 服务器错误等。
示例：
1. @app.errorhandler(404)
2. def page_not_found(error):
3. return "Page not found", 404
复制代码

@app.before_first_request()

作用：在应用处理第一个请求之前执行一次。适用于一些应用初始化的操作，例如数据库连接或缓存初始化等。
示例：
1. @app.before_first_request
2. def before_first_request():
3. print("This runs once before the first request.")
复制代码

@app.route() 支持 HTTP 方法的装饰器

作用：@app.route() 装饰器可以通过 methods 参数指定哪些 HTTP 方法（如 GET、POST、PUT、DELETE 等）可以触发该路由。
示例：
1. @app.route('/submit', methods=['POST'])
2. def submit():
3. return 'Form Submitted'
复制代码

2 修改路由以及服务器设置

2.1 基础设置

由于dify启动时会占用本地默认的 127.0.0.1:5000，为了避免冲突，我们就需要通过修改端口的形式来规避这个问题，用到的接口是：

app.run(debug=True, host='127.0.0.1', port=5001)
复制代码

app.run 中提供了修改基本信息的接口：

host：服务器的地址，window默认为 127.0.0.1
debug：调试模式是否启动
port：端口号。这里使用不同的端口号来分辨dify以及知识库服务器。

2.2test code

根据需求会post一个json的请求体

因此我们假设他传来的是json、调用get方法

from flask import Flask , request, jsonify
app = Flask(__name__)
@app.route('/retrieval',methods=['POST'])
def get_data():
data = request.get_json()
print(data)
return jsonify(data)
@app.route('/')
def default():
return 'hello'
if __name__ == '__main__':
app.run(debug=True, host='127.0.0.1', port=5001)
get_data()

复制代码

2.2.1 本地测试

先进行本地测试一下：

主页成功，测试 /retrieval页面：

问题不大，因为我们没有上传json文件，启动dify尝试一下
3 dify添加api测试

发现会报错，没办法访问：

3.1 问题解决：

问题思考

从计算机网络的角度来说，dify在WSL中运行，由于虚拟化，容器本地环境与windows的本地环境并不一致，即：当使用127.0.0.1进行访问时，访问的是容器内的主机，但我们的window环境并不在容器内部署，因此无法访问到window环境的127.0.0.1。中间需要一些NAT【单纯指网络地址转换】才能访问到主机

问题解决：找到了网上的一篇博主的推文：【docker知识】从容器中如何访问到宿主机里面提及了如何在容器内访问解释为主机的url

将API ENDPOINT改为：host.docker.internal
结果：

更换为docker能转换的url就能访问成功。

4 完善post类

根据api规范进行构造：

理论上是从 Dify_class-> Records开始构建的，但是依赖类需要写在前面，不用担心，这些都是基本功，不难的，就是复杂了一点，理清楚逻辑之后慢慢写就好：
PS：所有__repr__不要求写，我写着方便调试罢了
4.1 Dify_class 传入dify数据类

class Dify_class:
def __init__(self,posted_data:dict):
"""
dify有4个属性。
三个必填：知识库id、输入筛选器、检索设置（类）
一个选填：元数据信息（类）
:param posted_data: 收到的post，从json转换为字典形式
"""
self.knowledge_id:str = posted_data.get('knowledge_id')
self.query:str = posted_data.get('query')
self.retrieval_setting = Retrieval_setting(
posted_data.get('retrieval_setting')
)
self.metadata_condition = Metadata_condition(
posted_data.get('metadata_condition')
)
def __repr__(self):
res = f"knowledge_id:{self.knowledge_id} \nquery:{self.query} \n"f"{self.retrieval_setting.__repr__()}"
if self.metadata_condition != None:
res.join(self.metadata_condition.__repr__())
return res

复制代码

4.1.1 dify_class 两个依赖类

class Retrieval_setting:
def __init__(self, posted_data:dict):
self.top_k:int = posted_data.get('top_k')
self.score_threshold:float = posted_data.get('score_threshold')
def __repr__(self):
return f"\nretrieval_setting: \ntop_k:{self.top_k} \nscore_threshold:{self.score_threshold}"
class Metadata_condition:
def __init__(self, posted_data:dict):
if posted_data == None:
self.logical_operator = None
self.conditions = None
self.status = -1 # 用于查看有多少参数，用于repr， -1则为空，2则为都有（未完善）
else:
self.conditions = posted_data.get('conditions')
logical_operator_:str = posted_data.get('logical_operator')
if logical_operator_ != None:
self.logical_operator = logical_operator_
self.status = 2
else:
self.logical_operator = None
self.status = 1
def __repr__(self):
if self.status == -1:
return "None"
else:
return f'logical_operator:{self.logical_operator}\nconditions：{self.conditions}'

复制代码

4.2 record类

class Records:
def __init__(self,_content:str, _score:float, _title:str, _metadata:dict=None):
self.content = _content
self.score = _score
self.title = _title
self.metadata = Metadata(_metadata)
def to_dict(self):
"""
将record类转换为字典
:return: 返回单个字典类型的records
"""
res_dict = dict(
{
"metadata":{
"path":self.metadata.path,
"description":self.metadata.description
},
"score":self.score,
"title":self.title,
"content":self.content
}
)
return res_dict
def __repr__(self):
#没写metadata的
return f'*************\nscore:{self.score} \ncontent:{self.content} \ntitle:{self.title} \n*************\n'

复制代码

4.2.1 record 依赖类

class Metadata:
def __init__(self, record_dict:dict=None):
if record_dict != None:
self.path = record_dict.get("path")
self.description = record_dict.get("description")
else:
self.path = None
self.description = None

复制代码

4.3 测试dify类

类main函数【用于测试】

test.json文件：

{
"knowledge_id": "your-knowledge-id",
"query": "你的问题",
"retrieval_setting":{
"top_k": 2,
"score_threshold": 0.5
}
}

复制代码

main：

if __name__ == '__main__':
import json
with open('test.json', mode='r',encoding='utf8') as fp:
data = json.load(fp)
dify_t = Dify_class(data)
print(dify_t)
test_record = Records("test_content", 1.0, "dify_test")
print(test_record)

复制代码

5 接入服务器连接

5.1 导入相关包

from flask import Flask , request, jsonify
import dify_class ,json
#dify_class是4中的文件名称

复制代码

5.2 设置服务器

app = Flask(__name__)
@app.route('/retrieval',methods=['POST'])
def get_data():
data = request.get_json() #获取请求的json数据
dify_t = dify_class.Dify_class(data) #初始化dify请求类
print(dify_t) #调试输出
res = []
for i in range(dify_t.retrieval_setting.top_k): #模拟 topk
res.append(
dify_class.Records("test_content", 1.0, "dify_test").to_dict() #测试回复类，构造一个请求类->返回他的字典形式->放入res列表中
)
res_dict = {
"records": res
}
json_res = json.dumps(res_dict)
return json_res, 200

复制代码

5.3 主函数

#outside knowledge id_0001
if __name__ == '__main__':
app.run(debug=True, host='127.0.0.1', port=5001)
get_data()

复制代码

5.4 测试

<ol>启动服务器

进行召回测试

终于是显示测试效果出来了。能够返回你测试的样例就说明成功了
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册