ragflow的官方文檔：

HTTP API 接口 |抹布流 --- HTTP API | RAGFlow

接著前文，我們已經創建了知識庫，那么如何才能使用它呢？

當然也是通過網絡API的形式去調用它。本文將講解兩種方式：

1. Dify調用
2. python源碼調用

---

目錄

1.創建API-Key

2.DIfy接入ragflow

2.1.確定ragflow的域名

2.2. 確定IP地址

2.3.拼接完整域名

2.4.填寫信息

3.聊天測試

4.Python連接ragflow

4.1.知識庫管理

4.1.1.創建知識庫

4.1.2.刪除知識庫?

4.1.3.列出知識庫

4.1.4.更新知識庫

4.2.知識庫文件管理

4.2.1.上傳文件

4.2.2.更新文件?

4.2.3.下載文件

4.2.4.列出文件

4.2.5.刪除文件

4.2.6.解析文件

4.2.7.停止解析

---

1.創建API-Key

獲取ragflow知識庫ID：

2.DIfy接入ragflow

看看官網如何講解：External Knowledge API - Dify Docs

貌似找不出什么，因為這個節點肯定是由外部知識庫定義，只要遵循Dify的端點定義要求就行了，即：url路徑中要有一個/retrieval

2.1.確定ragflow的域名

接口是對外提供服務的，是后端服務，所以，他是ragflow-server提供的，我們可以看到，ragflow的容器中，ragflow-server是對外提供服務的，有兩種服務：

• 一個是web網頁端口的服務：默認是80端口，為了避免和dify端口沖突，我這里改為了8080端口，還有就是ssl端口443，我這里也改為了4434端口
• 另一個就是接口服務的端口：9380，這個就是我們的知識庫接口對外的服務端口，web前端通過接口提供的服務，與后端進行數據交互

?到這里，ragflow的外部知識庫接口前面的節點就確定了：http://{你的IP}:9380

2.2. 確定IP地址

由于我們的項目是通過docker啟動的，所以，統一使用 host.docker.internal:9380

這里解釋一下這個域名：

host.docker.internal 是Docker提供的一種方便的機制，用于在開發和測試環境中從容器訪問宿主機的服務。這個名稱在Docker Desktop for Windows和Docker Desktop for Mac上是可用的，它允許容器中的應用程序訪問宿主機上的服務和端口。

在不同的操作系統（如Windows和Mac）上，宿主機的IP地址可能會有所不同.host.docker.internal 提供了一個統一的名稱，使得容器中的配置在不同平臺上保持一致。

2.3.拼接完整域名

按照前面一步步推理，理論上來說，一個完整的api節點應該完整了，但是這就夠了嗎？

我們看下官方文檔：

發現每個接口都有個公共前綴：/api/v1

這其實也是絕大多數后端服務的通行做法，加上前面我們拼接處的api節點，現在才算是完整的拼接出了api節點了：http://host.docker.internal:9380/api/v1/dify/retrieval

我們現在可以確定Dify中關于外部知識庫的節點定義了，如下圖：

http://host.docker.internal:9380/api/v1/difyhttp://host.docker.internal:9380/api/v1/dify

2.4.填寫信息

如果配置正確會顯示出來。然后鏈接：

填寫剛剛的外部知識庫 API

3.聊天測試

創建一個聊天助手測試一下知識庫，

?

4.Python連接ragflow

Python 應用程序接口 |抹布流 --- Python API | RAGFlow

先安裝所需要的包：

pip install ragflow-sdk?

4.1.知識庫管理

4.1.1.創建知識庫

RAGFlow.create_dataset(name: str,avatar: Optional[str] = None,description: Optional[str] = None,embedding_model: Optional[str] = "BAAI/bge-large-zh-v1.5@BAAI",permission: str = "me", chunk_method: str = "naive",pagerank: int = 0,parser_config: DataSet.ParserConfig = None
) -> DataSet

??參數??	??類型??	??說明??	??可選值/默認值??
??name?（必填）?	str	數據集的唯一名稱（最長128字符，不區分大小寫）	-
??avatar??	str	頭像的Base64編碼	默認：None
??description??	str	數據集的簡要描述	默認：None
??permission??	str	數據集訪問權限	"me"（默認，僅自己可管理），"team"（全體團隊成員可管理）
??chunk_method??	str	數據集內容的分塊方法	"naive"（默認常規分塊）、"manual"（手動）、"qa"（問答）、"table"（表格）、"paper"（論文）、"book"（書籍）、"laws"（法律）、"presentation"（演示文稿）、"picture"（圖片）、"one"（單塊）、"email"（郵件）、"knowledge-graph"（知識圖譜）
??pagerank??	int	數據集的PageRank值（影響排序權重）	默認：0
??parser_config??	dict	解析器配置（根據chunk_method動態變化）	詳見下方說明

---

??parser_config 詳細說明（按分塊方法）??

??chunk_method??	??parser_config 配置??
"naive"（常規）	{"chunk_token_num":128, "delimiter":"\\n", "html4excel":False, "layout_recognize":True, "raptor":{"use_raptor":False}}
"qa"（問答）	{"raptor": {"use_raptor": False}}
"manual"（手動）	同上
"table"（表格）	None
"paper"（論文）	同"qa"
"book"（書籍）	同"qa"
"laws"（法律）	同"qa"
"picture"（圖片）	None
"presentation"（演示）	同"qa"
"one"（單塊）	None
"knowledge-graph"（知識圖譜）	{"chunk_token_num":128, "delimiter":"\\n", "entity_types":["organization","person","location","event","time"]}
"email"（郵件）	None

簡單的例子：?

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="***", base_url="http://host.docker.internal:9380")
#創建知識庫
dataset = rag_object.create_dataset(name="gyp")

在ragflow網站可以查看到：?

4.1.2.刪除知識庫?

RAGFlow.delete_datasets(ids: list[str] | None = None)

按 ID 刪除數據集。

如果為 None，則將刪除所有數據集。

id：8b025f18453711f0906f0242ac140004?

#8b025f18453711f0906f0242ac140004
rag_object.delete_datasets(ids=["8b025f18453711f0906f0242ac140004"])

4.1.3.列出知識庫

rag_object.list_datasets(page: int = 1, #頁碼page_size: int = 30,#每頁數量orderby: str = "create_time",#排序字段desc: bool = True,#是否降序id: str = None,#idname: str = None#名稱
) -> list[DataSet]

4.1.4.更新知識庫

DataSet.update(update_message: dict)

??參數??	??類型??	??說明??	??約束/可選值??
??update_message??	dict[str, str|int]	包含待更新屬性的字典，支持以下鍵：	-
→?"name"	str	數據集的新名稱	- 僅限基本多語言平面（BMP）字符<br>- 最長128字符<br>- 不區分大小寫
→?"avatar"	str	頭像的Base64編碼（Body參數）	- 最長65535字符
→?"embedding_model"	str	更新的嵌入模型名稱（Body參數）	- 格式必須為?model_name@model_factory<br>- 最長255字符<br>- 需確保?chunk_count=0?才能更新
→?"permission"	str	數據集權限	"me"（默認，僅自己可管理）<br>"team"（全體團隊成員可管理）
→?"pagerank"	int	數據集的PageRank值（影響排序權重）	- 默認值：0<br>- 范圍：0?~?100
→?"chunk_method"	enum<string>	數據集內容的分塊方法	"naive"（默認常規）<br>"book"（書籍）<br>"email"（郵件）<br>"laws"（法律）<br>"manual"（手動）<br>"one"（單塊）<br>"paper"（論文）<br>"picture"（圖片）<br>"presentation"（演示文稿）<br>"qa"（問答）<br>"table"（表格）<br>"tag"（標簽）

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(name="kb_name")
dataset = dataset[0]
dataset.update({"embedding_model":"BAAI/bge-zh-v1.5", "chunk_method":"manual"})

4.2.知識庫文件管理

4.2.1.上傳文件

DataSet.upload_documents(document_list: list[dict])#上傳文件

dataset = rag_object.create_dataset(name="kb_name")
dataset.upload_documents([{"display_name": "1.txt", "blob": "<BINARY_CONTENT_OF_THE_DOC>"}, {"display_name": "2.pdf", "blob": "<BINARY_CONTENT_OF_THE_DOC>"}])

“display_name”：（可選）要在數據集中顯示的文件名。

“blob”：（可選）要上傳的文件的二進制內容。

4.2.2.更新文件?

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(id='id')
dataset = dataset[0]
doc = dataset.list_documents(id="wdfxb5t547d")
doc = doc[0]
doc.update([{"parser_config": {"chunk_token_count": 256}}, {"chunk_method": "manual"}])

4.2.3.下載文件

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(id="id")
dataset = dataset[0]
doc = dataset.list_documents(id="wdfxb5t547d")
doc = doc[0]
open("~/ragflow.txt", "wb+").write(doc.download())
print(doc)

4.2.4.列出文件

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.create_dataset(name="kb_1")filename1 = "~/ragflow.txt"
blob = open(filename1 , "rb").read()
dataset.upload_documents([{"name":filename1,"blob":blob}])
for doc in dataset.list_documents(keywords="rag", page=0, page_size=12):print(doc)

4.2.5.刪除文件

from ragflow_sdk import RAGFlowrag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(name="kb_1")
dataset = dataset[0]
dataset.delete_documents(ids=["id_1","id_2"])

4.2.6.解析文件

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.create_dataset(name="dataset_name")
documents = [{'display_name': 'test1.txt', 'blob': open('./test_data/test1.txt',"rb").read()},{'display_name': 'test2.txt', 'blob': open('./test_data/test2.txt',"rb").read()},{'display_name': 'test3.txt', 'blob': open('./test_data/test3.txt',"rb").read()}
]
dataset.upload_documents(documents)
documents = dataset.list_documents(keywords="test")
ids = []
for document in documents:ids.append(document.id)
dataset.async_parse_documents(ids)
print("Async bulk parsing initiated.")

4.2.7.停止解析

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.create_dataset(name="dataset_name")
documents = [{'display_name': 'test1.txt', 'blob': open('./test_data/test1.txt',"rb").read()},{'display_name': 'test2.txt', 'blob': open('./test_data/test2.txt',"rb").read()},{'display_name': 'test3.txt', 'blob': open('./test_data/test3.txt',"rb").read()}
]
dataset.upload_documents(documents)
documents = dataset.list_documents(keywords="test")
ids = []
for document in documents:ids.append(document.id)
dataset.async_parse_documents(ids)
print("Async bulk parsing initiated.")
dataset.async_cancel_parse_documents(ids)
print("Async bulk parsing cancelled.")