跳到主要内容
版本: DEV

Python API

RAGFlow Python API 的完整参考。在继续之前，请确保您已准备好 RAGFlow API 密钥用于身份验证。

注意

运行以下命令以下载 Python SDK

pip install ragflow-sdk

错误码​

代码消息描述
400错误请求请求参数无效
401未经授权未经授权的访问
403禁止访问访问被拒绝
404未找到资源未找到
500内部服务器错误服务器内部错误
1001无效的块 ID无效的块 ID
1002块更新失败块更新失败

OpenAI 兼容 API​

创建聊天补全​

通过 OpenAI API 为给定的历史聊天对话创建模型响应。

参数​

model: str, *必填*​

用于生成响应的模型。服务器将自动解析此参数，因此目前您可以将其设置为任何值。

messages: list[object], *必填*​

用于生成响应的历史聊天消息列表。此列表必须至少包含一条角色为 user 的消息。

stream: boolean​

是否以流式接收响应。如果您希望一次性接收完整响应而不是流式接收，请明确将其设置为 false。

返回值​

  • 成功: 类似于 OpenAI 的响应消息
  • 失败: Exception

示例​

from openai import OpenAI

model = "model"
client = OpenAI(api_key="ragflow-api-key", base_url=f"http://ragflow_address/api/v1/chats_openai/<chat_id>")

completion = client.chat.completions.create(
    model=model,
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Who are you?"},
    ],
    stream=True
)

stream = True
if stream:
    for chunk in completion:
        print(chunk)
else:
    print(completion.choices[0].message.content)

数据集管理​

创建数据集​

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​

指定谁可以访问要创建的数据集。可用选项

  • "me": (默认) 只有您可以管理数据集。
  • "team": 所有团队成员都可以管理数据集。
chunk_method, str​

要创建的数据集的块方法。可用选项

  • "naive": 通用 (默认)
  • "manual: 手动
  • "qa": 问答
  • "table": 表格
  • "paper": 论文
  • "book": 书籍
  • "laws": 法律
  • "presentation": 演示文稿
  • "picture": 图片
  • "one": 单个文件
  • "email": 电子邮件
pagerank, int​

要创建的数据集的 PageRank。默认为 0。

parser_config​

数据集的解析器配置。ParserConfig 对象的属性根据选择的 chunk_method 而异

  • chunk_method="naive"
    {"chunk_token_num":128,"delimiter":"\\n","html4excel":False,"layout_recognize":True,"raptor":{"use_raptor":False}}.
  • chunk_method="qa"
    {"raptor": {"use_raptor": False}}
  • chunk_method="manual"
    {"raptor": {"use_raptor": False}}
  • chunk_method="table"
    None
  • chunk_method="paper"
    {"raptor": {"use_raptor": False}}
  • chunk_method="book"
    {"raptor": {"use_raptor": False}}
  • chunk_method="laws"
    {"raptor": {"use_raptor": False}}
  • chunk_method="picture"
    None
  • chunk_method="presentation"
    {"raptor": {"use_raptor": False}}
  • chunk_method="one"
    None
  • chunk_method="knowledge-graph"
    {"chunk_token_num":128,"delimiter":"\\n","entity_types":["organization","person","location","event","time"]}
  • chunk_method="email"
    None

返回值​

  • 成功: 一个 dataset 对象。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.create_dataset(name="kb_1")

删除数据集​

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

通过 ID 删除数据集。

参数​

ids: list[str] 或 None, *必填*​

要删除的数据集的 ID。默认为 None。

  • 如果为 None，则所有数据集将被删除。
  • 如果是一个 ID 数组，则只删除指定的数据集。
  • 如果是一个空数组，则不删除任何数据集。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

rag_object.delete_datasets(ids=["d94a8dc02c9711f0930f7fbc369eab6d","e94a8dc02c9711f0930f7fbc369eab6e"])

列出数据集​

RAGFlow.list_datasets(
    page: int = 1, 
    page_size: int = 30, 
    orderby: str = "create_time", 
    desc: bool = True,
    id: str = None,
    name: str = None
) -> list[DataSet]

列出数据集。

参数​

page: int​

指定显示数据集的页码。默认为 1。

page_size: int​

每页数据集的数量。默认为 30。

orderby: str​

数据集应按哪个字段排序。可用选项

  • "create_time" (默认)
  • "update_time"
desc: bool​

指示检索到的数据集是否按降序排序。默认为 True。

id: str​

要检索的数据集的 ID。默认为 None。

name: str​

要检索的数据集的名称。默认为 None。

返回值​

  • 成功: 一个 DataSet 对象列表。
  • 失败: Exception。

示例​

列出所有数据集​
for dataset in rag_object.list_datasets():
    print(dataset)
按 ID 检索数据集​
dataset = rag_object.list_datasets(id = "id_1")
print(dataset[0])

更新数据集​

DataSet.update(update_message: dict)

更新当前数据集的配置。

参数​

update_message: dict[str, str|int], *必填*​

一个表示要更新的属性的字典，包含以下键

  • "name": str 数据集的修订名称。
    • 仅限基本多文种平面 (BMP)
    • 最多 128 个字符
    • 不区分大小写
  • "avatar": (*请求体参数*), string​
    更新后的头像 Base64 编码。
    • 最多 65535 个字符
  • "embedding_model": (*请求体参数*), string​
    更新后的嵌入模型名称。
    • 更新 "embedding_model" 之前，请确保 "chunk_count" 为 0。
    • 最多 255 个字符
    • 必须遵循 model_name@model_factory 格式
  • "permission": (*请求体参数*), string​
    更新后的数据集权限。可用选项
    • "me": (默认) 只有您可以管理数据集。
    • "team": 所有团队成员都可以管理数据集。
  • "pagerank": (*请求体参数*), int​
    参考 设置 PageRank
    • 默认值: 0
    • 最小值: 0
    • 最大值: 100
  • "chunk_method": (*请求体参数*), enum<string>​
    数据集的块方法。可用选项
    • "naive": 通用 (默认)
    • "book": 书籍
    • "email": 电子邮件
    • "laws": 法律
    • "manual": 手动
    • "one": 单个文件
    • "paper": 论文
    • "picture": 图片
    • "presentation": 演示文稿
    • "qa": 问答
    • "table": 表格
    • "tag": 标签

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_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"})

数据集内的文件管理​

上传文档​

DataSet.upload_documents(document_list: list[dict])

将文档上传到当前数据集。

参数​

document_list: list[dict], *必填*​

一个字典列表，表示要上传的文档，每个字典包含以下键

  • "display_name": (可选) 在数据集中显示的文件名。
  • "blob": (可选) 要上传文件的二进制内容。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

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>"}])

更新文档​

Document.update(update_message:dict)

更新当前文档的配置。

参数​

update_message: dict[str, str|dict[]], *必填*​

一个表示要更新的属性的字典，包含以下键

  • "display_name": str 要更新的文档名称。
  • "meta_fields": dict[str, Any] 文档的元字段。
  • "chunk_method": str 应用于文档的解析方法。
    • "naive": 通用
    • "manual: 手动
    • "qa": 问答
    • "table": 表格
    • "paper": 论文
    • "book": 书籍
    • "laws": 法律
    • "presentation": 演示文稿
    • "picture": 图片
    • "one": 单个文件
    • "email": 电子邮件
  • "parser_config": dict[str, Any] 文档的解析配置。其属性根据选择的 "chunk_method" 而异
    • "chunk_method"="naive"
      {"chunk_token_num":128,"delimiter":"\\n","html4excel":False,"layout_recognize":True,"raptor":{"use_raptor":False}}.
    • chunk_method="qa"
      {"raptor": {"use_raptor": False}}
    • chunk_method="manual"
      {"raptor": {"use_raptor": False}}
    • chunk_method="table"
      None
    • chunk_method="paper"
      {"raptor": {"use_raptor": False}}
    • chunk_method="book"
      {"raptor": {"use_raptor": False}}
    • chunk_method="laws"
      {"raptor": {"use_raptor": False}}
    • chunk_method="presentation"
      {"raptor": {"use_raptor": False}}
    • chunk_method="picture"
      None
    • chunk_method="one"
      None
    • chunk_method="knowledge-graph"
      {"chunk_token_num":128,"delimiter":"\\n","entity_types":["organization","person","location","event","time"]}
    • chunk_method="email"
      None

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_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"}])

下载文档​

Document.download() -> bytes

下载当前文档。

返回值​

下载的文档内容（字节）。

示例​

from ragflow_sdk import RAGFlow

rag_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)

列出文档​

Dataset.list_documents(id:str =None, keywords: str=None, page: int=1, page_size:int = 30, order_by:str = "create_time", desc: bool = True) -> list[Document]

列出当前数据集中的文档。

参数​

id: str​

要检索的文档 ID。默认为 None。

keywords: str​

用于匹配文档标题的关键词。默认为 None。

page: int​

指定显示文档的页码。默认为 1。

page_size: int​

每页最大文档数量。默认为 30。

orderby: str​

文档应按哪个字段排序。可用选项

  • "create_time" (默认)
  • "update_time"
desc: bool​

指示检索到的文档是否按降序排序。默认为 True。

返回值​

  • 成功: 一个 Document 对象列表。
  • 失败: Exception。

一个 Document 对象包含以下属性

  • id: 文档 ID。默认为 ""。
  • name: 文档名称。默认为 ""。
  • thumbnail: 文档的缩略图图像。默认为 None。
  • dataset_id: 与文档关联的数据集 ID。默认为 None。
  • chunk_method 块方法名称。默认为 "naive"。
  • source_type: 文档的来源类型。默认为 "local"。
  • type: 文档的类型或类别。默认为 ""。保留供将来使用。
  • created_by: str 文档创建者。默认为 ""。
  • size: int 文档大小（字节）。默认为 0。
  • token_count: int 文档中的 token 数量。默认为 0。
  • chunk_count: int 文档中的块数量。默认为 0。
  • progress: float 当前处理进度（百分比）。默认为 0.0。
  • progress_msg: str 指示当前进度状态的消息。默认为 ""。
  • process_begin_at: datetime 文档处理开始时间。默认为 None。
  • process_duation: float 处理持续时间（秒）。默认为 0.0。
  • run: str 文档的处理状态
    • "UNSTART" (默认)
    • "RUNNING" (运行中)
    • "CANCEL" (取消)
    • "DONE" (完成)
    • "FAIL" (失败)
  • status: str 保留供将来使用。
  • parser_config: ParserConfig 解析器的配置对象。其属性根据选择的 chunk_method 而异
    • chunk_method="naive"
      {"chunk_token_num":128,"delimiter":"\\n","html4excel":False,"layout_recognize":True,"raptor":{"use_raptor":False}}.
    • chunk_method="qa"
      {"raptor": {"use_raptor": False}}
    • chunk_method="manual"
      {"raptor": {"use_raptor": False}}
    • chunk_method="table"
      None
    • chunk_method="paper"
      {"raptor": {"use_raptor": False}}
    • chunk_method="book"
      {"raptor": {"use_raptor": False}}
    • chunk_method="laws"
      {"raptor": {"use_raptor": False}}
    • chunk_method="presentation"
      {"raptor": {"use_raptor": False}}
    • chunk_method="picture"
      None
    • chunk_method="one"
      None
    • chunk_method="email"
      None

示例​

from ragflow_sdk import RAGFlow

rag_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)

删除文档​

DataSet.delete_documents(ids: list[str] = None)

通过 ID 删除文档。

参数​

ids: list[list]​

要删除的文档 ID。默认为 None。如果未指定，则将删除数据集中的所有文档。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_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"])

解析文档​

DataSet.async_parse_documents(document_ids:list[str]) -> None

解析当前数据集中的文档。

参数​

document_ids: list[str], *必填*​

要解析的文档 ID。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

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(document_ids:list[str])-> None

停止解析指定的文档。

参数​

document_ids: list[str], *必填*​

应停止解析的文档 ID。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

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.")

数据集内的块管理​

添加块​

Document.add_chunk(content:str, important_keywords:list[str] = []) -> Chunk

向当前文档添加一个块。

参数​

content: str, *必填*​

块的文本内容。

important_keywords: list[str]​

与块关联的关键术语或短语。

返回值​

  • 成功: 一个 Chunk 对象。
  • 失败: Exception。

一个 Chunk 对象包含以下属性

  • id: str: 块 ID。
  • content: str 块的文本内容。
  • important_keywords: list[str] 与块关联的关键术语或短语列表。
  • create_time: str 块创建（添加到文档）的时间。
  • create_timestamp: float 表示块创建时间的 Unix 时间戳（自 1970 年 1 月 1 日以来的秒数）。
  • dataset_id: str 关联数据集的 ID。
  • document_name: str 关联文档的名称。
  • document_id: str 关联文档的 ID。
  • available: bool 块在数据集中的可用状态。值选项
    • False: 不可用
    • True: 可用 (默认)

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
datasets = rag_object.list_datasets(id="123")
dataset = datasets[0]
doc = dataset.list_documents(id="wdfxb5t547d")
doc = doc[0]
chunk = doc.add_chunk(content="xxxxxxx")

列出块​

Document.list_chunks(keywords: str = None, page: int = 1, page_size: int = 30, id : str = None) -> list[Chunk]

列出当前文档中的块。

参数​

keywords: str​

用于匹配块内容的关键词。默认为 None

page: int​

指定显示块的页码。默认为 1。

page_size: int​

每页最大块数量。默认为 30。

id: str​

要检索的块 ID。默认值: None

返回值​

  • 成功: 一个 Chunk 对象列表。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets("123")
dataset = dataset[0]
docs = dataset.list_documents(keywords="test", page=1, page_size=12)
for chunk in docs[0].list_chunks(keywords="rag", page=0, page_size=12):
    print(chunk)

删除块​

Document.delete_chunks(chunk_ids: list[str])

通过 ID 删除块。

参数​

chunk_ids: list[str]​

要删除的块 ID。默认为 None。如果未指定，将删除当前文档的所有块。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(id="123")
dataset = dataset[0]
doc = dataset.list_documents(id="wdfxb5t547d")
doc = doc[0]
chunk = doc.add_chunk(content="xxxxxxx")
doc.delete_chunks(["id_1","id_2"])

更新块​

Chunk.update(update_message: dict)

更新当前块的内容或配置。

参数​

update_message: dict[str, str|list[str]|int] *必填*​

一个表示要更新的属性的字典，包含以下键

  • "content": str 块的文本内容。
  • "important_keywords": list[str] 与块关联的关键术语或短语列表。
  • "available": bool 块在数据集中的可用状态。值选项
    • False: 不可用
    • True: 可用 (默认)

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(id="123")
dataset = dataset[0]
doc = dataset.list_documents(id="wdfxb5t547d")
doc = doc[0]
chunk = doc.add_chunk(content="xxxxxxx")
chunk.update({"content":"sdfx..."})

检索块​

RAGFlow.retrieve(question:str="", dataset_ids:list[str]=None, document_ids=list[str]=None, page:int=1, page_size:int=30, similarity_threshold:float=0.2, vector_similarity_weight:float=0.3, top_k:int=1024,rerank_id:str=None,keyword:bool=False,highlight:bool=False) -> list[Chunk]

从指定数据集中检索块。

参数​

question: str, *必填*​

用户查询或查询关键词。默认为 ""。

dataset_ids: list[str], *必填*​

要搜索的数据集 ID。默认为 None。

document_ids: list[str]​

要搜索的文档 ID。默认为 None。必须确保所有选定的文档使用相同的嵌入模型，否则将发生错误。

page: int​

检索文档的起始索引。默认为 1。

page_size: int​

要检索的最大块数量。默认为 30。

similarity_threshold: float​

最小相似度分数。默认为 0.2。

vector_similarity_weight: float​

向量余弦相似度的权重。默认为 0.3。如果 x 表示向量余弦相似度，则 (1 - x) 是词项相似度的权重。

top_k: int​

参与向量余弦计算的块数量。默认为 1024。

rerank_id: str​

重排模型的 ID。默认为 None。

keyword: bool​

指示是否启用基于关键词的匹配

  • True: 启用基于关键词的匹配。
  • False: 禁用基于关键词的匹配 (默认)。
highlight: bool​

指定是否在结果中启用匹配词高亮显示

  • True: 启用匹配词高亮显示。
  • False: 禁用匹配词高亮显示 (默认)。

返回值​

  • 成功: 一个 Chunk 对象列表，表示文档块。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.list_datasets(name="ragflow")
dataset = dataset[0]
name = 'ragflow_test.txt'
path = './test_data/ragflow_test.txt'
documents =[{"display_name":"test_retrieve_chunks.txt","blob":open(path, "rb").read()}]
docs = dataset.upload_documents(documents)
doc = docs[0]
doc.add_chunk(content="This is a chunk addition test")
for c in rag_object.retrieve(dataset_ids=[dataset.id],document_ids=[doc.id]):
  print(c)

聊天助手管理​

创建聊天助手​

RAGFlow.create_chat(
    name: str, 
    avatar: str = "", 
    dataset_ids: list[str] = [], 
    llm: Chat.LLM = None, 
    prompt: Chat.Prompt = None
) -> Chat

创建一个聊天助手。

参数​

name: str, *必填*​

聊天助手的名称。

avatar: str​

头像的 Base64 编码。默认为 ""。

dataset_ids: list[str]​

关联数据集的 ID。默认为 [""]。

llm: Chat.LLM​

要创建的聊天助手的 LLM 设置。默认为 None。当值为 None 时，将生成一个包含以下默认值的字典。一个 LLM 对象包含以下属性

  • model_name: str
    聊天模型名称。如果为 None，将使用用户的默认聊天模型。
  • temperature: float
    控制模型预测的随机性。较低的 temperature 会产生更保守的响应，而较高的 temperature 会产生更具创意和多样性的响应。默认为 0.1。
  • top_p: float
    也称为“核采样”，此参数设置一个阈值，用于选择一小组词汇进行采样。它侧重于最可能的词汇，排除不太可能的词汇。默认为 0.3
  • presence_penalty: float
    通过惩罚已在对话中出现的词汇来阻止模型重复相同的信息。默认为 0.2。
  • frequency penalty: float
    类似于 presence penalty，这减少了模型频繁重复相同词汇的倾向。默认为 0.7。
prompt: Chat.Prompt​

LLM 需要遵循的指令。一个 Prompt 对象包含以下属性

  • similarity_threshold: float RAGFlow 在检索时采用加权关键词相似度和加权向量余弦相似度的组合，或加权关键词相似度和加权重排分数的组合。如果相似度分数低于此阈值，相应的块将从结果中排除。默认值为 0.2。
  • keywords_similarity_weight: float 此参数设置关键词相似度在与向量余弦相似度或重排模型相似度的混合相似度分数中的权重。通过调整此权重，您可以控制关键词相似度相对于其他相似度度量的影响。默认值为 0.7。
  • top_n: int 此参数指定相似度分数高于 similarity_threshold 的前 N 个块的数量，这些块将被送入 LLM。LLM *仅*访问这些“前 N 个”块。默认值为 8。
  • variables: list[dict[]] 此参数列出了在**聊天配置**的“系统”字段中使用的变量。请注意
    • knowledge 是一个保留变量，表示检索到的块。
    • “系统”中的所有变量都应该用花括号括起来。
    • 默认值为 [{"key": "knowledge", "optional": True}]。
  • rerank_model: str 如果未指定，将使用向量余弦相似度；否则，将使用重排分数。默认为 ""。
  • top_k: int 指的是根据特定排名标准从列表或集合中重新排序或选择前 k 个项的过程。默认为 1024。
  • empty_response: str 如果数据集中没有检索到与用户问题相关的内容，将使用此作为响应。如果希望 LLM 在没有找到内容时自由发挥，请留空。默认为 None。
  • opener: str 对用户的开场问候语。默认为 "您好！我是您的助手，有什么可以帮助您的吗？"。
  • show_quote: bool 指示是否显示文本来源。默认为 True。
  • prompt: str 提示内容。

返回值​

  • 成功: 一个 Chat 对象，表示聊天助手。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
datasets = rag_object.list_datasets(name="kb_1")
dataset_ids = []
for dataset in datasets:
    dataset_ids.append(dataset.id)
assistant = rag_object.create_chat("Miss R", dataset_ids=dataset_ids)

更新聊天助手​

Chat.update(update_message: dict)

更新当前聊天助手的配置。

参数​

update_message: dict[str, str|list[str]|dict[]], *必填*​

一个表示要更新的属性的字典，包含以下键

  • "name": str 聊天助手的修订名称。
  • "avatar": str 头像的 Base64 编码。默认为 ""
  • "dataset_ids": list[str] 要更新的数据集。
  • "llm": dict LLM 设置
    • "model_name": str 聊天模型名称。
    • "temperature": float 控制模型预测的随机性。较低的 temperature 会产生更保守的响应，而较高的 temperature 会产生更具创意和多样性的响应。
    • "top_p": float 也称为“核采样”，此参数设置一个阈值，用于选择一小组词汇进行采样。
    • "presence_penalty": float 通过惩罚已在对话中出现的词汇来阻止模型重复相同的信息。
    • "frequency penalty": float 类似于 presence penalty，这减少了模型频繁重复相同词汇的倾向。
  • "prompt" : LLM 需要遵循的指令。
    • "similarity_threshold": float RAGFlow 在检索时采用加权关键词相似度和加权向量余弦相似度的组合，或加权关键词相似度和加权重排分数的组合。此参数设置用户查询与块之间的相似度阈值。如果相似度分数低于此阈值，相应的块将从结果中排除。默认值为 0.2。
    • "keywords_similarity_weight": float 此参数设置关键词相似度在与向量余弦相似度或重排模型相似度的混合相似度分数中的权重。通过调整此权重，您可以控制关键词相似度相对于其他相似度度量的影响。默认值为 0.7。
    • "top_n": int 此参数指定相似度分数高于 similarity_threshold 的前 N 个块的数量，这些块将被送入 LLM。LLM *仅*访问这些“前 N 个”块。默认值为 8。
    • "variables": list[dict[]] 此参数列出了在**聊天配置**的“系统”字段中使用的变量。请注意
      • knowledge 是一个保留变量，表示检索到的块。
      • “系统”中的所有变量都应该用花括号括起来。
      • 默认值为 [{"key": "knowledge", "optional": True}]。
    • "rerank_model": str 如果未指定，将使用向量余弦相似度；否则，将使用重排分数。默认为 ""。
    • "empty_response": str 如果数据集中没有检索到与用户问题相关的内容，将使用此作为响应。如果希望 LLM 在没有检索到内容时自由发挥，请留空。默认为 None。
    • "opener": str 对用户的开场问候语。默认为 "您好！我是您的助手，有什么可以帮助您的吗？"。
    • "show_quote: bool 指示是否显示文本来源。默认为 True。
    • "prompt": str 提示内容。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
datasets = rag_object.list_datasets(name="kb_1")
dataset_id = datasets[0].id
assistant = rag_object.create_chat("Miss R", dataset_ids=[dataset_id])
assistant.update({"name": "Stefan", "llm": {"temperature": 0.8}, "prompt": {"top_n": 8}})

删除聊天助手​

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

通过 ID 删除聊天助手。

参数​

ids: list[str]​

要删除的聊天助手 ID。默认为 None。如果为空或未指定，将删除系统中的所有聊天助手。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.delete_chats(ids=["id_1","id_2"])

列出聊天助手​

RAGFlow.list_chats(
    page: int = 1, 
    page_size: int = 30, 
    orderby: str = "create_time", 
    desc: bool = True,
    id: str = None,
    name: str = None
) -> list[Chat]

列出聊天助手。

参数​

page: int​

指定显示聊天助手的页码。默认为 1。

page_size: int​

每页聊天助手的数量。默认为 30。

orderby: str​

结果按哪个属性排序。可用选项

  • "create_time" (默认)
  • "update_time"
desc: bool​

指示检索到的聊天助手是否按降序排序。默认为 True。

id: str​

要检索的聊天助手的 ID。默认为 None。

name: str​

要检索的聊天助手的名称。默认为 None。

返回值​

  • 成功: 一个 Chat 对象列表。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
for assistant in rag_object.list_chats():
    print(assistant)

会话管理​

创建与聊天助手的会话​

Chat.create_session(name: str = "New session") -> Session

创建与当前聊天助手的新会话。

参数​

name: str​

要创建的聊天会话名称。

返回值​

  • 成功: 一个 Session 对象，包含以下属性
    • id: str 创建会话的自动生成的唯一标识符。
    • name: str 创建会话的名称。
    • message: list[Message] 创建会话的开场消息。默认值: [{"role": "assistant", "content": "您好！我是您的助手，有什么可以帮助您的吗？"}]
    • chat_id: str 关联聊天助手的 ID。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
assistant = rag_object.list_chats(name="Miss R")
assistant = assistant[0]
session = assistant.create_session()

更新聊天助手的会话​

Session.update(update_message: dict)

更新当前聊天助手的当前会话。

参数​

update_message: dict[str, Any], *必填*​

一个表示要更新的属性的字典，只有一个键

  • "name": str 会话的修订名称。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
assistant = rag_object.list_chats(name="Miss R")
assistant = assistant[0]
session = assistant.create_session("session_name")
session.update({"name": "updated_name"})

列出聊天助手的会话​

Chat.list_sessions(
    page: int = 1, 
    page_size: int = 30, 
    orderby: str = "create_time", 
    desc: bool = True,
    id: str = None,
    name: str = None
) -> list[Session]

列出与当前聊天助手关联的会话。

参数​

page: int​

指定显示会话的页码。默认为 1。

page_size: int​

每页会话的数量。默认为 30。

orderby: str​

会话应按哪个字段排序。可用选项

  • "create_time" (默认)
  • "update_time"
desc: bool​

指示检索到的会话是否按降序排序。默认为 True。

id: str​

要检索的聊天会话的 ID。默认为 None。

name: str​

要检索的聊天会话的名称。默认为 None。

返回值​

  • 成功: 一个与当前聊天助手关联的 Session 对象列表。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
assistant = rag_object.list_chats(name="Miss R")
assistant = assistant[0]
for session in assistant.list_sessions():
    print(session)

删除聊天助手的会话​

Chat.delete_sessions(ids:list[str] = None)

通过 ID 删除当前聊天助手的会话。

参数​

ids: list[str]​

要删除的会话 ID。默认为 None。如果未指定，将删除与当前聊天助手关联的所有会话。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
assistant = rag_object.list_chats(name="Miss R")
assistant = assistant[0]
assistant.delete_sessions(ids=["id_1","id_2"])

与聊天助手对话​

Session.ask(question: str = "", stream: bool = False, **kwargs) -> Optional[Message, iter[Message]]

向指定的聊天助手提问以开始 AI 对话。

注意

在流式模式下，并非所有响应都包含参考，这取决于系统的判断。

参数​

question: str, *必填*​

开始 AI 对话的问题。默认为 ""

stream: bool​

指示是否以流式方式输出响应

  • True: 启用流式输出 (默认)。
  • False: 禁用流式输出。
**kwargs​

prompt (system) 中的参数。

返回值​

  • 如果 stream 设置为 False，则返回包含问题响应的 Message 对象。
  • 如果 stream 设置为 True，则返回包含多个 message 对象的迭代器 (iter[Message])。

以下显示 Message 对象的属性

id: str​

自动生成的消息 ID。

content: str​

消息内容。默认为 "您好！我是您的助手，有什么可以帮助您的吗？"。

reference: list[Chunk]​

一个 Chunk 对象列表，表示消息的参考，每个对象包含以下属性

  • id str
    块 ID。
  • content str
    块的内容。
  • img_id str
    块快照的 ID。仅当块源为图片、PPT、PPTX 或 PDF 文件时适用。
  • document_id str
    引用文档的 ID。
  • document_name str
    引用文档的名称。
  • position list[str]
    块在引用文档中的位置信息。
  • dataset_id str
    引用文档所属数据集的 ID。
  • similarity float
    块的综合相似度分数，范围从 0 到 1，值越高表示相似度越高。它是 vector_similarity 和 term_similarity 的加权和。
  • vector_similarity float
    块的向量相似度分数，范围从 0 到 1，值越高表示向量嵌入之间的相似度越高。
  • term_similarity float
    块的关键词相似度分数，范围从 0 到 1，值越高表示关键词之间的相似度越高。

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
assistant = rag_object.list_chats(name="Miss R")
assistant = assistant[0]
session = assistant.create_session()    

print("\n==================== Miss R =====================\n")
print("Hello. What can I do for you?")

while True:
    question = input("\n==================== User =====================\n> ")
    print("\n==================== Miss R =====================\n")
    
    cont = ""
    for ans in session.ask(question, stream=True):
        print(ans.content[len(cont):], end='', flush=True)
        cont = ans.content

创建与 Agent 的会话​

Agent.create_session(**kwargs) -> Session

创建与当前 Agent 的会话。

参数​

**kwargs​

begin 组件中的参数。

返回值​

  • 成功: 一个 Session 对象，包含以下属性
    • id: str 创建会话的自动生成的唯一标识符。
    • message: list[Message] 创建会话助手的消息。默认值: [{"role": "assistant", "content": "您好！我是您的助手，有什么可以帮助您的吗？"}]
    • agent_id: str 关联 Agent 的 ID。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow, Agent

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
agent_id = "AGENT_ID"
agent = rag_object.list_agents(id = agent_id)[0]
session = agent.create_session()

与 Agent 对话​

Session.ask(question: str="", stream: bool = False) -> Optional[Message, iter[Message]]

向指定的 Agent 提问以开始 AI 对话。

注意

在流式模式下，并非所有响应都包含参考，这取决于系统的判断。

参数​

question: str​

开始 AI 对话的问题。如果 **Begin** 组件接受参数，则不需要问题。

stream: bool​

指示是否以流式方式输出响应

  • True: 启用流式输出 (默认)。
  • False: 禁用流式输出。

返回值​

  • 如果 stream 设置为 False，则返回包含问题响应的 Message 对象
  • 如果 stream 设置为 True，则返回包含多个 message 对象的迭代器 (iter[Message])。

以下显示 Message 对象的属性

id: str​

自动生成的消息 ID。

content: str​

消息内容。默认为 "您好！我是您的助手，有什么可以帮助您的吗？"。

reference: list[Chunk]​

一个 Chunk 对象列表，表示消息的参考，每个对象包含以下属性

  • id str
    块 ID。
  • content str
    块的内容。
  • image_id str
    块快照的 ID。仅当块源为图片、PPT、PPTX 或 PDF 文件时适用。
  • document_id str
    引用文档的 ID。
  • document_name str
    引用文档的名称。
  • position list[str]
    块在引用文档中的位置信息。
  • dataset_id str
    引用文档所属数据集的 ID。
  • similarity float
    块的综合相似度分数，范围从 0 到 1，值越高表示相似度越高。它是 vector_similarity 和 term_similarity 的加权和。
  • vector_similarity float
    块的向量相似度分数，范围从 0 到 1，值越高表示向量嵌入之间的相似度越高。
  • term_similarity float
    块的关键词相似度分数，范围从 0 到 1，值越高表示关键词之间的相似度越高。

示例​

from ragflow_sdk import RAGFlow, Agent

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
AGENT_id = "AGENT_ID"
agent = rag_object.list_agents(id = AGENT_id)[0]
session = agent.create_session()    

print("\n===== Miss R ====\n")
print("Hello. What can I do for you?")

while True:
    question = input("\n===== User ====\n> ")
    print("\n==== Miss R ====\n")
    
    cont = ""
    for ans in session.ask(question, stream=True):
        print(ans.content[len(cont):], end='', flush=True)
        cont = ans.content

列出 Agent 会话​

Agent.list_sessions(
    page: int = 1, 
    page_size: int = 30, 
    orderby: str = "update_time", 
    desc: bool = True,
    id: str = None
) -> List[Session]

列出与当前 Agent 关联的会话。

参数​

page: int​

指定显示会话的页码。默认为 1。

page_size: int​

每页会话的数量。默认为 30。

orderby: str​

会话应按哪个字段排序。可用选项

  • "create_time"
  • "update_time"(默认)
desc: bool​

指示检索到的会话是否按降序排序。默认为 True。

id: str​

要检索的 Agent 会话的 ID。默认为 None。

返回值​

  • 成功: 一个与当前 Agent 关联的 Session 对象列表。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
AGENT_id = "AGENT_ID"
agent = rag_object.list_agents(id = AGENT_id)[0]
sessons = agent.list_sessions()
for session in sessions:
    print(session)

删除 Agent 的会话​

Agent.delete_sessions(ids: list[str] = None)

通过 ID 删除 Agent 的会话。

参数​

ids: list[str]​

要删除的会话 ID。默认为 None。如果未指定，将删除与 Agent 关联的所有会话。

返回值​

  • 成功: 不返回任何值。
  • 失败: Exception

示例​

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
AGENT_id = "AGENT_ID"
agent = rag_object.list_agents(id = AGENT_id)[0]
agent.delete_sessions(ids=["id_1","id_2"])

Agent 管理​

列出 Agent​

RAGFlow.list_agents(
    page: int = 1, 
    page_size: int = 30, 
    orderby: str = "create_time", 
    desc: bool = True,
    id: str = None,
    title: str = None
) -> List[Agent]

列出 Agent。

参数​

page: int​

指定显示 Agent 的页码。默认为 1。

page_size: int​

每页 Agent 的数量。默认为 30。

orderby: str​

结果按哪个属性排序。可用选项

  • "create_time" (默认)
  • "update_time"
desc: bool​

指示检索到的 Agent 是否按降序排序。默认为 True。

id: str​

要检索的 Agent 的 ID。默认为 None。

name: str​

要检索的 Agent 的名称。默认为 None。

返回值​

  • 成功: 一个 Agent 对象列表。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
for agent in rag_object.list_agents():
    print(agent)

创建 Agent​

RAGFlow.create_agent(
    title: str,
    dsl: dict,
    description: str | None = None
) -> None

创建一个 Agent。

参数​

title: str​

指定 Agent 的标题。

dsl: dict​

指定 Agent 的画布 DSL。

description: str​

Agent 的描述。默认为 None。

返回值​

  • 成功: 无返回值。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.create_agent(
  title="Test Agent",
  description="A test agent",
  dsl={
    # ... canvas DSL here ...
  }
)

更新 Agent​

RAGFlow.update_agent(
    agent_id: str,
    title: str | None = None,
    description: str | None = None,
    dsl: dict | None = None
) -> None

更新一个 Agent。

参数​

agent_id: str​

指定要更新的 Agent 的 ID。

title: str​

指定 Agent 的新标题。如果不想更新，则为 None。

dsl: dict​

指定 Agent 的新画布 DSL。如果不想更新，则为 None。

description: str​

Agent 的新描述。如果不想更新，则为 None。

返回值​

  • 成功: 无返回值。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.update_agent(
  agent_id="58af890a2a8911f0a71a11b922ed82d6",
  title="Test Agent",
  description="A test agent",
  dsl={
    # ... canvas DSL here ...
  }
)

删除 Agent​

RAGFlow.delete_agent(
    agent_id: str
) -> None

删除一个 Agent。

参数​

agent_id: str​

指定要删除的 Agent 的 ID。

返回值​

  • 成功: 无返回值。
  • 失败: Exception。

示例​

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.delete_agent("58af890a2a8911f0a71a11b922ed82d6")