尝试dify自定义知识库

根据官网教程，可以从知识库的右上角外部知识库进行添加外部知识库

前往 “知识库” 页，点击右上角的 “外部知识库 API”，轻点 “添加外部知识库 API”。

按照页面提示，依次填写以下内容：

  • 知识库的名称，允许自定义名称，用于区分所连接的不同外部知识 API；

  • API 接口地址，外部知识库的连接地址，示例 api-endpoint/retrieval；详细说明请参考外部知识库 API；

  • API Key，外部知识库连接密钥，详细说明请参考外部知识库 API；

    image-20250326204900832

因为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 服务器错误等。

  • 示例：

    
     
    
         
    		@app.errorhandler(404)
    
         
    		def page_not_found(error):
    
         
    		    return "Page not found", 404
@app.before_first_request()

  • 作用：在应用处理第一个请求之前执行一次。适用于一些应用初始化的操作，例如数据库连接或缓存初始化等。

  • 示例：

    
     
    
         
    		@app.before_first_request
    
         
    		def before_first_request():
    
         
    		    print("This runs once before the first request.")
@app.route() 支持 HTTP 方法的装饰器

  • 作用：@app.route() 装饰器可以通过 methods 参数指定哪些 HTTP 方法（如 GET、POST、PUT、DELETE 等）可以触发该路由。

  • 示例：

    
     
    
         
    		@app.route('/submit', methods=['POST'])
    
         
    		def submit():
    
         
    		    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 中提供了修改基本信息的接口：

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

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

    image-20250328161849930

    因此我们假设他传来的是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 本地测试

先进行本地测试一下：

image-20250328162114712

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

image-20250328162159024

image-20250328162215346

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

3 dify添加api测试

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

image-20250328163025929

3.1 问题解决：
  • 问题思考

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

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

    将API ENDPOINT改为：host.docker.internal

    image-20250328170817389

  • 结果：

    image-20250328170628821

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

4 完善post类

根据api规范进行构造：

image-20250328173748099

理论上是从 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 测试

  1. 启动服务器

    image-20250329173751319

  2. 进行召回测试

    image-20250329173820464

    终于是显示测试效果出来了。能够返回你测试的样例就说明成功了 😭 😭，后续就是根据他post的东西来进行检索了。

原创作者: io-T-T 转载于: https://www.cnblogs.com/io-T-T/p/18865007
Logo
EazyDevelop社区

一站式 AI 云服务平台

更多推荐

cover

5分钟搞定！MySQL/PostgreSQL 到 Elasticsearch 的实时同步

avatar EazyDevelop社区
cover

国内低代码平台：2025 年国内主流平台盘点

avatar EazyDevelop社区
cover

两小时，我搭了一套销售提成计算系统

avatar EazyDevelop社区