Python API

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

提示

运行以下命令下载 Python SDK：

pip install ragflow-sdk

错误码 (ERROR CODES)

状态码	消息 (Message)	描述 (Description)
400	Bad Request	请求参数无效
401	Unauthorized	未授权的访问
403	Forbidden	访问被拒绝
404	Not Found	未找到资源
500	Internal Server Error	服务器内部错误
1001	Invalid Chunk ID	无效的块 ID
1002	Chunk Update Failed	块更新失败

兼容 OpenAI 的 API (OpenAI-Compatible API)

创建对话补全 (Create chat completion)

通过 OpenAI 的 API，为给定的历史对话生成模型回复。

参数 (Parameters)

chat_id: `string`, 必填

已有的对话助手 ID。此值是请求路径的一部分：/api/v1/openai/<chat_id>/chat/completions。

model: `string`, 必填

用于生成回复的模型。您也可以使用遗留的占位符值 "model" 以继续使用该对话助手所配置的模型。

messages: `list[object]`, 必填

用于生成回复的历史对话消息列表。该列表必须至少包含一条角色为 user 的消息。

stream: `boolean`

是否以流式（Stream）接收回复。如果您希望一次性接收完整的回复而不是流式回复，请显式将此值设置为 false。

返回值 (Returns)

成功：类似于 OpenAI 的响应消息 (message)
失败：Exception

示例 (Examples)

from openai import OpenAI
import json

model = "glm-4-flash@ZHIPU-AI"
client = OpenAI(api_key="ragflow-api-key", base_url="http://ragflow_address/api/v1/openai/<chat_id>/chat")

stream = True
reference = True

request_kwargs = dict(
    model=model,
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Who are you?"},
        {"role": "assistant", "content": "I am an AI assistant named..."},
        {"role": "user", "content": "Can you tell me how to install neovim"},
    ],
    extra_body={
        "reference": reference,
        "reference_metadata": {
            "include": True,
            "fields": ["author", "year", "source"],
        },
    },
)

if stream:
    completion = client.chat.completions.create(stream=True, **request_kwargs)
    for chunk in completion:
        print(chunk)
else:
    resp = client.chat.completions.with_raw_response.create(
        stream=False, **request_kwargs
    )
    print("status:", resp.http_response.status_code)
    raw_text = resp.http_response.text
    print("raw:", raw_text)

    data = json.loads(raw_text)
    print("assistant:", data["choices"][0]["message"].get("content"))
    print("reference:", data["choices"][0]["message"].get("reference"))

当 extra_body.reference 为 true 时，流式返回的最终块可能包含 choices[0].delta.reference，而非流式返回可能会包含 choices[0].message.reference。

当 extra_body.reference_metadata.include 为 true 时，在流式和非流式回复中，每个引用的块都可能包含一个 document_metadata 对象。

数据集管理 (DATASET MANAGEMENT)

创建数据集 (Create dataset)

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",
    parser_config: DataSet.ParserConfig = None
) -> DataSet

创建一个数据集（数据集/知识库）。

参数 (Parameters)

name: `string`, 必填

要创建的数据集的唯一名称。它必须符合以下要求：

最多 128 个字符。
不区分大小写。

avatar: `string`

头像的 Base64 编码。默认值为 None。

description: `string`

要创建的数据集的简短描述。默认值为 None。

permission

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

"me"：（默认）仅自己可以管理该数据集。
"team"：所有团队成员都可以管理该数据集。

chunk_method, `string`

要创建的数据集的分块方法 (Chunking Method)。可用选项：

"naive"：通用（默认）
"manual"：手动
"qa"：问答 (Q&A)
"table"：表格
"paper"：论文
"book"：图书
"laws"：法律
"presentation"：演示文稿 (PPT)
"picture"：图片
"one"：单文档
"email"：电子邮件

parser_config

数据集的解析器配置 (Parser Configuration)。一个 ParserConfig 对象的属性取决于所选的分块方法 chunk_method：

chunk_method="naive"：
{"chunk_token_num":512,"delimiter":"\\n","html4excel":False,"layout_recognize":True,"raptor":{"use_raptor":False},"parent_child":{"use_parent_child":False,"children_delimiter":"\\n"}}。
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

返回值 (Returns)

成功：一个 DataSet 对象。
失败：Exception

示例 (Examples)

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

删除数据集 (Delete datasets)

RAGFlow.delete_datasets(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除数据集。

参数 (Parameters)

ids: `list[str]` 或 `None`

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

如果省略、或者设置为 null/空数组，则不删除任何数据集。
如果提供了 ID 数组，则仅删除与这些 ID 匹配的数据集。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前用户拥有的所有数据集。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

列出数据集 (List datasets)

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

获取数据集列表。

参数 (Parameters)

page: `int`

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

page_size: `int`

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

orderby: `string`

对数据集进行排序的字段。可用选项：

"create_time"（默认）
"update_time"

desc: `bool`

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

id: `string`

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

name: `string`

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

include_parsing_status: `bool`

是否在每个返回的 DataSet 对象中包含文档解析状态的统计信息。默认值为 False。当设置为 True 时，每个 DataSet 对象将包含以下附加属性：

unstart_count: int 尚未开始解析的文档数量。
running_count: int 当前正在解析的文档数量。
cancel_count: int 已取消解析的文档数量。
done_count: int 已成功解析的文档数量。
fail_count: int 解析失败的文档数量。

返回值 (Returns)

成功：DataSet 对象列表。
失败：Exception。

示例 (Examples)

列出所有数据集

for dataset in rag_object.list_datasets():
    print(dataset)

根据 ID 检索数据集

dataset = rag_object.list_datasets(id = "id_1")
print(dataset[0])

列出包含解析状态的数据集

for dataset in rag_object.list_datasets(include_parsing_status=True):
    print(dataset.done_count, dataset.fail_count, dataset.running_count)

更新数据集 (Update dataset)

DataSet.update(update_message: dict)

更新当前数据集的配置。

参数 (Parameters)

update_message: `dict[str, str|int]`, 必填

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

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

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

数据集内文件管理 (FILE MANAGEMENT WITHIN DATASET)

上传文档 (Upload documents)

DataSet.upload_documents(document_list: list[dict])

上传文档到当前数据集。

参数 (Parameters)

document_list: `list[dict]`, 必填

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

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

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

更新文档 (Update document)

Document.update(update_message:dict)

更新当前文档的配置。

参数 (Parameters)

update_message: `dict[str, str|dict[]]`, 必填

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

"display_name": string 要更新的文档的名称。
"meta_fields": dict[str, Any] 文档的元数据字段。
"chunk_method": string 应用于文档的解析方法。
- "naive"：通用
- "manual"：手动
- "qa"：问答 (Q&A)
- "table"：表格
- "paper"：论文
- "book"：图书
- "laws"：法律
- "presentation"：演示文稿 (PPT)
- "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},"parent_child":{"use_parent_child":False,"children_delimiter":"\\n"}}。
- 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

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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_num": 256}}, {"chunk_method": "manual"}])

下载文档 (Download document)

Document.download() -> bytes

下载当前文档。

返回值 (Returns)

以字节（Bytes）格式返回下载的文档。

示例 (Examples)

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)

列出文档 (List documents)

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

获取当前数据集中的文档列表。

参数 (Parameters)

id: `string`

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

keywords: `string`

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

page: `int`

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

page_size: `int`

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

orderby: `string`

对文档进行排序的字段。可用选项：

"create_time"（默认）
"update_time"

desc: `bool`

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

create_time_from: `int`

Unix 时间戳，用于过滤在此时间之后创建的文档。0 表示无过滤。默认值为 0。

create_time_to: `int`

Unix 时间戳，用于过滤在此时间之前创建的文档。0 表示无过滤。默认值为 0。

返回值 (Returns)

成功：Document 对象列表。
失败：Exception。

一个 Document 对象包含以下属性：

id: 文档 ID。默认值为 ""。
name: 文档名称。默认值为 ""。
thumbnail: 文档缩略图。默认值为 None。
dataset_id: 与该文档关联的数据集 ID。默认值为 None。
chunk_method: 分块方法名称。默认值为 "naive"。
source_type: 文档的来源类型。默认值为 "local"。
type: 文档的类型或分类。默认值为 ""。保留供未来使用。
created_by: string 文档的创建者。默认值为 ""。
size: int 文档的大小（以字节为单位）。默认值为 0。
token_count: int 文档中的 Token 数量。默认值为 0。
chunk_count: int 文档中的分块数量。默认值为 0。
progress: float 当前处理进度的百分比。默认值为 0.0。
progress_msg: string 表示当前进度状态的消息。默认值为 ""。
process_begin_at: datetime 文档处理的开始时间。默认值为 None。
process_duration: float 处理持续时间（以秒为单位）。默认值为 0.0。
run: string 文档的处理状态：
- "UNSTART"（默认）
- "RUNNING"
- "CANCEL"
- "DONE"
- "FAIL"
status: string 保留供未来使用。
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

示例 (Examples)

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)

删除文档 (Delete documents)

DataSet.delete_documents(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除文档。

参数 (Parameters)

ids: `list[str]` 或 `None`

要删除的文档的 ID 列表。默认值为 None。

如果省略、或者设置为 null/空数组，则不删除任何文档。
如果提供了 ID 数组，则仅删除与这些 ID 匹配的文档。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前数据集中的所有文档。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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.delete_documents(delete_all=True)

解析文档 (Parse documents)

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

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

参数 (Parameters)

document_ids: `list[str]`, 必填

要解析的文档的 ID 列表。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

解析文档（包含文档状态） (Parse documents with status)

DataSet.parse_documents(document_ids: list[str]) -> list[tuple[str, str, int, int]]

以异步方式解析当前数据集中的文档。

此方法封装了 async_parse_documents()。它会等待所有解析任务完成后再返回详细结果，包括每个文档的解析状态和统计信息。如果发生键盘中断（例如 Ctrl+C），所有未决的解析任务都将被平滑取消。

参数 (Parameters)

document_ids: `list[str]`, 必填

要解析的文档的 ID 列表。

返回值 (Returns)

包含详细解析结果的元组列表：

[
  (document_id: str, status: str, chunk_count: int, token_count: int),
  ...
]

status：最终解析状态（例如 success、failed、cancelled）。
chunk_count：从该文档生成的块数量。
token_count：已处理的 Token 总数。

示例 (Example)

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
dataset = rag_object.create_dataset(name="dataset_name")
documents = dataset.list_documents(keywords="test")
ids = [doc.id for doc in documents]

try:
    finished = dataset.parse_documents(ids)
    for doc_id, status, chunk_count, token_count in finished:
        print(f"Document {doc_id} parsing finished with status: {status}, chunks: {chunk_count}, tokens: {token_count}")
except KeyboardInterrupt:
    print("\nParsing interrupted by user. All pending tasks have been cancelled.")
except Exception as e:
    print(f"Parsing failed: {e}")

停止解析文档 (Stop parsing documents)

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

停止解析指定的文档。

参数 (Parameters)

document_ids: `list[str]`, 必填

应停止解析的文档的 ID 列表。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

数据集内分块管理 (CHUNK MANAGEMENT WITHIN DATASET)

添加块 (Add chunk)

Document.add_chunk(content:str, important_keywords:list[str] = [], questions:list[str] = [], image_base64:str = None, *, tag_kwd:list[str] = []) -> Chunk

向当前文档中添加一个块 (Chunk)。

参数 (Parameters)

content: `string`, 必填

块的文本内容。

important_keywords: `list[str]`

用于标记块的关键术语或短语列表。

questions: `list[str]`

在嵌入块时使用的可选问题列表。

image_base64: `string`

与该块关联的 Base64 编码的图像。如果该块已包含图像，新图像将垂直拼接在现有图像下方。

tag_kwd: `list[str]`

与该块关联的标签关键词。

返回值 (Returns)

成功：一个 Chunk 对象。
失败：Exception。

一个 Chunk 对象包含以下属性：

id: string：块 ID。
content: string 块的文本内容。
important_keywords: list[str] 标记在块上的关键术语或短语列表。
tag_kwd: list[str] 与块关联的标签关键词列表。
questions: list[str] 与块关联的问题列表。
image_id: string 与块关联的图像 ID（若无图像则为空字符串）。
create_time: string 块的创建时间（添加到文档的时间）。
create_timestamp: float 块的创建时间戳，以秒为单位（自 1970 年 1 月 1 日算起）。
dataset_id: string 关联的数据集 ID。
document_name: string 关联的文档名称。
document_id: string 关联的文档 ID。
available: bool 块在数据集中的可用状态。可用选项：
- False：不可用
- True：（默认）可用

示例 (Examples)

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

添加包含图像的块：

import base64

with open("image.jpg", "rb") as f:
    img_b64 = base64.b64encode(f.read()).decode()
chunk = doc.add_chunk(content="description of image", image_base64=img_b64)

列出块 (List chunks)

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

获取当前文档中的块列表。

参数 (Parameters)

keywords: `string`

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

page: `int`

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

page_size: `int`

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

id: `string`

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

返回值 (Returns)

成功：Chunk 对象列表。
失败：Exception。

示例 (Examples)

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)

删除块 (Delete chunks)

Document.delete_chunks(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除块。

参数 (Parameters)

ids: `list[str]` 或 `None`

要删除的块的 ID 列表。默认值为 None。

如果省略、或者设置为 null/空数组，则不删除任何块。
If 提供了 ID 数组，则仅删除与这些 ID 匹配的块。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前文档中的所有块。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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"])
doc.delete_chunks(delete_all=True)

更新块 (Update chunk)

Chunk.update(update_message: dict)

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

参数 (Parameters)

update_message: `dict[str, str|list[str]|bool]` 必填

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

"content": string 块的文本内容。
"important_keywords": list[str] 用于标记块的关键术语或短语列表。
"questions": list[str] 与块关联的问题列表。
"tag_kwd": list[str] 与该块关联的标签关键词列表。
"positions": list 块在源文档中更新后的位置信息。
"available": bool 块在数据集中的可用状态。可用选项：
- False：不可用
- True：（默认）可用
"image_base64": string 与块关联的 Base64 编码的图像内容。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

检索块 (Retrieve chunks)

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,cross_languages:list[str]=None,metadata_condition: dict=None) -> list[Chunk]

从指定的数据集检索相关块。

参数 (Parameters)

question: `string`, 必填

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

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) 为术语词频相似度 (Term Similarity) 权重。

top_k: `int`

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

rerank_id: `string`

重排模型 (Rerank Model) ID。默认值为 None。

keyword: `bool`

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

True：启用基于关键词的匹配。
False：（默认）禁用基于关键词的匹配。

cross_languages: `list[string]`

要翻译的目标语言列表，以便进行多语言的关键词检索。

metadata_condition: `dict`

针对 meta_fields 的过滤条件。

返回值 (Returns)

成功：表示文档块的 Chunk 对象列表。
失败：Exception

示例 (Examples)

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)

对话助手管理 (CHAT ASSISTANT MANAGEMENT)

创建对话助手 (Create chat assistant)

RAGFlow.create_chat(
    name: str,
    icon: str = "",
    dataset_ids: list[str] | None = None,
    llm_id: str | None = None,
    llm_setting: dict | None = None,
    prompt_config: dict | None = None,
    **kwargs
) -> Chat

创建一个对话助手。

参数 (Parameters)

name: `string`, 必填

对话助手的名称。

icon: `string`

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

dataset_ids: `list[str]`

关联的数据集 ID 列表。默认值为 []。省略或为空时，SDK 会创建一个空的对话助手，您可以在稍后绑定数据集。

llm_id: `str | None`

要使用的 LLM 模型名称/ID。如果为 None，则使用用户的默认对话模型。默认值为 None。

llm_setting: `dict | None`

LLM 生成参数的配置。默认值为 None（此时应用服务器端默认配置）。支持的键值：

"temperature": float 控制模型输出的随机性。值越高，输出越有创意；值越低，回复越确定。默认值为 0.1。
"top_p": float 设置核采样阈值。模型仅考虑累计概率达到 top_p 的 Token。默认值为 0.3。
"presence_penalty": float 根据 Token 到目前为止是否在文本中出现进行惩罚，增加模型讨论新话题的可能性。默认值为 0.4。
"frequency_penalty": float 根据 Token 在文本中的现有频率进行惩罚，降低模型重复相同内容的可能性。默认值为 0.7。
"max_token": int 回复中生成的最大 Token 数量。默认值为 512。

prompt_config: `dict | None`

针对 LLM 的提示词行为配置。默认值为 None（此时应用服务器端默认配置）。支持的键值：

"system": string 核心系统提示词 (System Prompt)，用于定义助手的角色。
"empty_response": string 在未检索到任何相关信息时返回的特定消息。若留空，LLM 将会自行生成回复。默认值为 None。
"prologue": string 向用户展示的开场白问候语。默认值为 "Hi! I’m your assistant. What can I do for you?"。
"quote": boolean 决定助手是否应在回复中包含引用或来源信息。默认值为 True。
"parameters": list[dict] 系统提示词中使用的变量列表。每个条目必须包含一个 "key" (string) 和一个 "optional" (boolean) 状态。knowledge 键保留用于检索到的上下文块。默认值：[{"key": "knowledge", "optional": true}]。

返回值 (Returns)

成功：代表对话助手的 Chat 对象。
失败：Exception

示例 (Examples)

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)

更新对话助手 (Update chat assistant)

Chat.update(update_message: dict)

对当前对话助手的配置设置进行部分更新。

Chat.update() 访问 PATCH /api/v1/chats/{chat_id} 接口。仅修改指定的键，其他现有字段均予以保留。

参数 (Parameters)

update_message: `dict`, 必填

包含要更新的属性的字典。支持的键包含：

"name": string 对话助手的新名称。
"icon": string 代表助手头像的 Base64 编码字符串。
"dataset_ids": list[string] 与助手关联的数据集 ID 列表。
"llm_id": string 要使用的 LLM 的名称或 ID。
"llm_setting": dict LLM 生成参数的配置：
- "temperature": float 控制模型输出的随机性。
- "top_p": float 设置核采样阈值。
- "presence_penalty": float 根据 Token 是否已在文本中出现进行惩罚。
- "frequency_penalty": float 根据 Token 在文本中的现有频率进行惩罚。
- "max_token": int 回复中生成的最大 Token 数。
"prompt_config": dict LLM 的提示词配置：
- "system": string 定义助手角色的核心系统提示词。
- "empty_response": string 未检索到相关信息时返回的消息。留空则允许 LLM 即兴发挥。
- "prologue": string 展示给用户的开场白问候语。
- "quote": boolean 决定助手是否应包含引用或来源参考。
- "parameters": list[dict] 系统提示词中使用的变量（例如保留的 knowledge 键）。
"similarity_threshold": float 检索到的上下文块所需的最小相似度分数。默认值为 0.2。
"vector_similarity_weight": float 混合检索评分中向量相似度的权重。默认值为 0.3。
"top_n": int 作为上下文提供给 LLM 的相关度排名前 N 的块数量。默认值为 6。
"top_k": int 检索用于重排的初始候选池大小。默认值为 1024。
"rerank_id": string 重排模型的唯一标识符。如果留空，则使用标准向量余弦相似度进行排序。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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_setting": {"temperature": 0.8}, "top_n": 8})

删除对话助手 (Delete chat assistants)

RAGFlow.delete_chats(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除对话助手。

参数 (Parameters)

ids: `list[str]` 或 `None`

要删除的对话助手的 ID 列表。默认值为 None。

如果省略、或者设置为 null/空数组，则不删除任何对话助手。
如果提供了 ID 数组，则仅删除与这些 ID 匹配的对话助手。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前用户拥有的所有对话助手。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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"])
rag_object.delete_chats(delete_all=True)

列出对话助手 (List chat assistants)

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

获取对话助手列表。

参数 (Parameters)

page: `int`

指定显示对话助手的页码。默认值为 1。

page_size: `int`

每页的对话助手数量。默认值为 30。

orderby: `string`

排序的属性。可用选项：

"create_time"（默认）
"update_time"

desc: `bool`

指示检索到的对话助手是否按降序排序。默认值为 True。

id: `string | None`

精确匹配的对话助手 ID。默认值为 None。

name: `string | None`

根据对话助手的精确名称进行过滤。默认值为 None。

keywords: `string | None`

对对话助手名称进行不区分大小写的模糊搜索。默认值为 None。

owner_ids: `string | list[string] | None`

根据一个或多个所有者租户 ID 过滤结果。默认值为 None。

parser_id: `string | None`

根据特定的解析器类型过滤结果。默认值为 None。

如果指定了 id 或 name，精确匹配过滤的优先级高于 keywords 提供的模糊匹配。

返回值 (Returns)

成功：Chat 对象列表。
失败：Exception。

示例 (Examples)

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)

会话管理 (SESSION MANAGEMENT)

创建与对话助手的会话 (Create session with chat assistant)

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

创建与当前对话助手的会话 (Session)。

参数 (Parameters)

name: `string`

要创建的对话会话的名称。

返回值 (Returns)

成功：包含以下属性的 Session 对象：
- id: string 自动生成的唯一会话 ID。
- name: string 会话名称。
- message: list[Message] 该会话的开场消息。默认值：[{"role": "assistant", "content": "Hi! I am your assistant, can I help you?"}]
- chat_id: string 关联的对话助手 ID。
失败：Exception

示例 (Examples)

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

更新对话会话 (Update chat assistant's session)

Session.update(update_message: dict)

更新当前对话助手的当前会话。

参数 (Parameters)

update_message: `dict[str, Any]`, 必填

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

"name": string 修改后的会话名称。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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

列出对话会话 (List chat assistant's sessions)

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

获取与当前对话助手关联的会话列表。

参数 (Parameters)

page: `int`

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

page_size: `int`

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

orderby: `string`

会话排序的字段。可用选项：

"create_time"（默认）
"update_time"

desc: `bool`

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

id: `string`

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

name: `string`

要检索的对话会话的名称。默认值为 None。

user_id: `str`

用于过滤会话的可选用户自定义 ID。默认值为 None。

返回值 (Returns)

成功：当前对话助手关联的 Session 对象列表。
失败：Exception。

示例 (Examples)

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)

删除对话会话 (Delete chat assistant's sessions)

Chat.delete_sessions(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除当前对话助手的会话。

参数 (Parameters)

ids: `list[str]` 或 `None`

要删除的会话的 ID 列表。默认值为 None。

如果省略、或者设置为 null/空数组，则不删除任何会话。
如果提供了 ID 数组，则仅删除与这些 ID 匹配的会话。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前对话助手的所有会话。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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"])
assistant.delete_sessions(delete_all=True)

与对话助手对话 (Converse with chat assistant)

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

向指定的对话助手提问以启动 AI 对话。

注意

在流式传输模式下，并非所有回复都包含引用（Reference），这取决于系统的判断。

参数 (Parameters)

question: `string`, 必填

开启 AI 对话的提问内容。默认值为 ""。

stream: `bool`

指示是否以流式输出回复：

True：（默认）启用流式传输。
False：禁用流式传输。

**kwargs

系统提示词 (System Prompt) 中的变量参数。

返回值 (Returns)

如果 stream 设置为 False，返回包含提问回复的 Message 对象。
如果 stream 设置为 True，返回包含多个 Message 对象的迭代器 (iter[Message])。

以下是 Message 对象的属性：

id: `string`

自动生成的消息 ID。

content: `string`

消息内容。默认值为 "Hi! I am your assistant, can I help you?"。

reference: `list[Chunk]`

代表该消息引用源的 Chunk 对象列表，每个对象包含以下属性：

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

示例 (Examples)

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

创建与智能体的会话 (Create session with agent)

Agent.create_session(**kwargs) -> Session

创建与当前智能体 (Agent) 的会话。

参数 (Parameters)

**kwargs

在 begin 组件中配置的参数。

另外支持：

release (bool | str, 可选)：当设置为 True (或 "true") 时，仅与已发布的智能体应用创建会话。

返回值 (Returns)

成功：包含以下属性的 Session 对象：
- id: string 自动生成的唯一会话 ID。
- message: list[Message] 创建的智能体助手的消息。默认值：[{"role": "assistant", "content": "Hi! I am your assistant, can I help you?"}]
- agent_id: string 关联的智能体 ID。
失败：Exception

示例 (Examples)

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.get_agent(agent_id)
session = agent.create_session()
# 或者在发布模式下创建：
# session = agent.create_session(release=True)

与智能体对话 (Converse with agent)

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

通过统一的补全接口向指定的智能体提问。

注意

在流式传输模式下，并非所有回复都包含引用（Reference），这取决于系统的判断。

参数 (Parameters)

question: `string`

发送给智能体的用户消息。如果 Begin 组件带参数，question 可以为空字符串。

stream: `bool`

指示是否以流式输出回复：

True：启用流式传输。
False：禁用流式传输。

kwargs: `dict`

转发给补全 API 的附加请求参数。常用选项：

inputs：在 Begin 组件中定义的变量。
session_id：继续现有的会话，而不是创建新会话。
release：使用智能体最新发布的版本。
return_trace：在回复中包含执行轨迹信息。
当前工作流支持的其他自定义 Begin 组件参数。

返回值 (Returns)

如果 stream 设置为 False，返回包含提问回复的 Message 对象。
如果 stream 设置为 True，返回包含多个 Message 对象的迭代器 (iter[Message])。

以下是 Message 对象的属性：

id: `string`

自动生成的消息 ID。

content: `string`

消息内容。默认值为 "Hi! I am your assistant, can I help you?"。

reference: `list[Chunk]`

代表该消息引用源的 Chunk 对象列表，每个对象包含以下属性：

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

示例 (Examples)

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.get_agent(AGENT_id)
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

使用 Begin 输入变量并请求追踪输出：

from ragflow_sdk import RAGFlow, Agent

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
agent = rag_object.get_agent("AGENT_ID")
session = agent.create_session()

message = session.ask(
    "",
    stream=False,
    inputs={
        "line_var": {
            "type": "line",
            "value": "I am line_var",
        }
    },
    return_trace=True,
)

print(message.content)
print(message.reference)

列出智能体会话 (List agent sessions)

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

获取与当前智能体关联的会话列表。

参数 (Parameters)

page: `int`

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

page_size: `int`

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

orderby: `string`

会话排序的字段。可用选项：

"create_time"
"update_time"（默认）

desc: `bool`

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

id: `string`

要检索的智能体会话的 ID。默认值为 None。

返回值 (Returns)

成功：与当前智能体关联的 Session 对象列表。
失败：Exception。

示例 (Examples)

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.get_agent(AGENT_id)
sessions = agent.list_sessions()
for session in sessions:
    print(session)

删除智能体会话 (Delete agent's sessions)

Agent.delete_sessions(ids: list[str] | None = None, delete_all: bool = False)

根据 ID 删除智能体的会话。

参数 (Parameters)

ids: `list[str]` 或 `None`

要删除的会话的 ID 列表。默认值为 None。

如果省略、或者设置为 None/空数组，则不删除任何会话。
如果提供了 ID 数组，则仅删除与这些 ID 匹配的会话。

delete_all: `bool`

当省略 ids 或将其设置为 None/空列表时，是否删除当前智能体的所有会话。默认值为 False。

返回值 (Returns)

成功：不返回值。
失败：Exception

示例 (Examples)

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.get_agent(AGENT_id)
agent.delete_sessions(ids=["id_1","id_2"])
agent.delete_sessions(delete_all=True)

智能体管理 (AGENT MANAGEMENT)

列出智能体 (List agents)

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

获取智能体列表。这是一个集合接口，始终返回一个列表。

参数 (Parameters)

page: `int`

指定显示智能体的页码。默认值为 1。

page_size: `int`

每页的智能体数量。默认值为 30。

orderby: `string`

结果排序的属性。可用选项：

"create_time"
"update_time"（默认）

desc: `bool`

指示检索到的智能体是否按降序排序。默认值为 True。

返回值 (Returns)

成功：Agent 对象列表。
失败：Exception。

示例 (Examples)

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)

获取智能体 (Get agent)

RAGFlow.get_agent(agent_id: str) -> Agent

根据 ID 获取单个智能体，并返回详细的智能体负载 (Payload)。

参数 (Parameters)

agent_id: `string`

要检索的智能体的 ID。

返回值 (Returns)

成功：一个 Agent 对象。
失败：Exception。

示例 (Examples)

from ragflow_sdk import RAGFlow

rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
agent = rag_object.get_agent("AGENT_ID")
print(agent)

创建智能体 (Create agent)

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

创建一个智能体。

参数 (Parameters)

title: `string`

指定智能体的标题。

dsl: `dict`

指定智能体画布的 DSL。

description: `string`

智能体的描述。默认值为 None。

返回值 (Returns)

成功：无返回值。
失败：Exception。

示例 (Examples)

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={
    # ... 此处为画布 DSL ...
  }
)

更新智能体 (Update agent)

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

更新一个智能体。

参数 (Parameters)

agent_id: `string`

指定要更新的智能体的 ID。

title: `string`

指定智能体的新标题。如果不打算更新此项，请传入 None。

dsl: `dict`

指定智能体的新画布 DSL。如果不打算更新此项，请传入 None。

description: `string`

智能体的新描述。如果不打算更新此项，请传入 None。

返回值 (Returns)

成功：无返回值。
失败：Exception。

示例 (Examples)

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={
    # ... 此处为画布 DSL ...
  }
)

删除智能体 (Delete agent)

RAGFlow.delete_agent(
    agent_id: str
) -> None

删除一个智能体。

参数 (Parameters)

agent_id: `string`

指定要删除的智能体的 ID。

返回值 (Returns)

成功：无返回值。
失败：Exception。

示例 (Examples)

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

记忆管理 (Memory Management)

创建记忆 (Create Memory)

RAGFlow.create_memory(
    name: str, 
    memory_type: list[str], 
    embd_id: str, 
    llm_id: str
) -> Memory

创建一条新记忆。

参数 (Parameters)

name: `string`, 必填

要创建的记忆的唯一名称。它必须符合以下要求：

仅限基本多语言平面 (BMP) 字符
最多 128 个字符

memory_type: `list[str]`, 必填

指定要提取的记忆类型。可用选项：

raw：用户和智能体之间的原始对话内容。默认必填。
semantic：关于用户和世界的通用知识和事实。
episodic：带时间戳的特定事件和体验记录。
procedural：学习到的技能、习惯以及自动化的程序。

embd_id: `string`, 必填

要使用的嵌入模型名称。例如："BAAI/bge-large-zh-v1.5@BAAI"

最多 255 个字符
必须遵循 model_name@model_factory 格式

llm_id: `string`, 必填

要使用的对话模型名称。例如："glm-4-flash@ZHIPU-AI"

最多 255 个字符
必须遵循 model_name@model_factory 格式

返回值 (Returns)

成功：一个 Memory 对象。
失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory = rag_object.create_memory("name", ["raw"], "BAAI/bge-large-zh-v1.5@SILICONFLOW", "glm-4-flash@ZHIPU-AI")

更新记忆 (Update Memory)

Memory.update(
	update_dict: dict
) -> Memory

更新指定记忆的配置。

参数 (Parameters)

update_dict: `dict`, 必填

要更新的配置。可用配置：

name: string, 可选

修改后的记忆名称。
- 仅限基本多语言平面 (BMP) 字符
- 最多 128 个字符，可选
avatar: string, 可选

更新后的头像 Base64 编码。
- 最多 65535 个字符
permission: enum<string>, 可选

更新后的记忆权限。可用选项：
- "me"：（默认）仅自己可以管理该记忆。
- "team"：所有团队成员都可以管理该记忆。
llm_id: string, 可选

要使用的对话模型名称。例如："glm-4-flash@ZHIPU-AI"
- 最多 255 个字符
- 必须遵循 model_name@model_factory 格式
description: string, 可选

记忆的描述。默认值为 None。
memory_size: int, 可选

默认值为 5*1024*1024 字节。计算公式为：每条消息的内容 + 其嵌入向量（≈ 内容 + 维度 × 8 字节）。例如：一条带有 1024 维嵌入的 1 KB 消息占用大约 9 KB。5 MB 的默认限制大约可以保存 500 条此类消息。
- 最大值：10 * 1024 * 1024 字节
forgetting_policy: enum<string>, 可选

当达到大小限制时，基于所选的策略驱逐现有数据，以释放新消息的空间。可用选项：
- "FIFO"：（默认）优先移除 forget_at 时间最早的消息。当设置了 forget_at 的消息池不足时，退而按 valid_at 的升序（最早的优先）选择消息进行清理。
temperature: (请求体参数), float, 可选

调整输出随机性。越低 = 越确定；越高 = 越有创意。
- 范围 [0, 1]
system_prompt: (请求体参数), string, 可选

为 AI 助手定义系统级的指令和角色。它由 memory/utils/prompt_util.py 中的 PromptAssembler 根据选定的 memory_type 自动装配。此提示词为整个对话奠定了基础行为和上下文。
- 请保持其中的 OUTPUT REQUIREMENTS 和 OUTPUT FORMAT 部分不变。
user_prompt: (请求体参数), string, 可选

代表用户的自定义设置，即 AI 需要直接响应的具体问题或指令。默认值为 None。

返回值 (Returns)

成功：一个 Memory 对象。
失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.update({"name": "New_name"})

列出记忆 (List Memory)

RAGFlow.list_memory(
    page: int = 1, 
    page_size: int = 50, 
    tenant_id: str | list[str] = None, 
    memory_type: str | list[str] = None, 
    storage_type: str = None, 
    keywords: str = None) -> dict

列出记忆。

参数 (Parameters)

page: `int`, 可选

指定显示记忆的页码。默认值为 1。

page_size: `int`, 可选

每页的记忆数量。默认值为 50。

tenant_id: `string` 或 `list[str]`, 可选

所有者的租户 ID，支持搜索多个 ID。

memory_type: `string` 或 `list[str]`, 可选

记忆的类型（与创建时设置一致）。如果记忆的类型包含在所提供的值中，则匹配成功。可用选项：

raw
semantic
episodic
procedural

storage_type: `string`, 可选

消息的存储格式。可用选项：

table：（默认）

keywords: `string`, 可选

要检索的记忆名称，支持模糊搜索。

返回值 (Returns)

成功：返回包含 Memory 对象列表和总数的字典。

{"memory_list": list[Memory], "total_count": int}

失败：Exception

示例 (Examples)

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

获取记忆配置 (Get Memory Config)

Memory.get_config()

获取指定记忆的配置。

参数 (Parameters)

无

返回值 (Returns)

成功：一个 Memory 对象。

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.get_config()

删除记忆 (Delete Memory)

RAGFlow.delete_memory(
    memory_id: str
) -> None

删除指定的记忆。

参数 (Parameters)

memory_id: `string`, 必填

该记忆的 ID。

返回值 (Returns)

成功：无返回值。

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.delete_memory("your memory_id")

列出记忆中的消息 (List messages of a memory)

Memory.list_memory_messages(
    agent_id: str | list[str]=None, 
    keywords: str=None, 
    page: int=1, 
    page_size: int=50
) -> dict

列出指定记忆中的消息。

参数 (Parameters)

agent_id: `string` 或 `list[str]`, 可选

根据来源智能体 ID 过滤消息。支持多个值。

keywords: `string`, 可选

根据会话 ID 过滤消息。此字段支持模糊搜索。

page: `int`, 可选

指定显示消息的页码。默认值为 1。

page_size: `int`, 可选

每页的消息数量。默认值为 50。

返回值 (Returns)

成功：包含消息和元信息的字典。

{"messages": {"message_list": [{message dict}], "total_count": int}, "storage_type": "table"}

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.list_memory_messages()

添加消息 (Add Message)

RAGFlow.add_message(
    memory_id: list[str], 
    agent_id: str, 
    session_id: str, 
    user_input: str, 
    agent_response: str, 
    user_id: str = ""
) -> str

向指定的记忆中添加消息。

参数 (Parameters)

memory_id: `list[str]`, 必填

保存消息的记忆 ID 列表。

agent_id: `string`, 必填

消息来源智能体的 ID。

session_id: `string`, 必填

消息会话的 ID。

user_input: `string`, 必填

用户提供的文本输入。

agent_response: `string`, 必填

AI 智能体生成的文本回复。

user_id: `string`, 可选

参与智能体对话的用户 ID。默认值为 ""。

返回值 (Returns)

成功：文本 "All add to task."

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
message_payload = {
    "memory_id": memory_ids,
    "agent_id": agent_id,
    "session_id": session_id,
    "user_id": "",
    "user_input": "Your question here",
    "agent_response": """
Your agent response here
"""
}
client.add_message(**message_payload)

遗忘消息 (Forget Message)

Memory.forget_message(message_id: int) -> bool

遗忘指定的消息。遗忘后，该消息将不会被智能体检索，并且会被遗忘策略优先进行清理。

参数 (Parameters)

message_id: `int`, 必填

要遗忘的消息的 ID。

返回值 (Returns)

成功：True

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.forget_message(message_id)

更新消息状态 (Update message status)

Memory.update_message_status(message_id: int, status: bool) -> bool

更新消息状态，启用或禁用该消息。一旦消息被禁用，它将不会被智能体检索。

参数 (Parameters)

message_id: `int`, 必填

要启用或禁用的消息的 ID。

status: `bool`, 必填

消息的状态。True = 已启用 (enabled)，False = 已禁用 (disabled)。

返回值 (Returns)

成功：True

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow, Memory
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.update_message_status(message_id, True)

搜索消息 (Search message)

RAGFlow.search_message(
    query: str, 
    memory_id: list[str], 
    agent_id: str=None, 
    session_id: str=None,
    user_id: str=None,
    similarity_threshold: float=0.2, 
    keywords_similarity_weight: float=0.7, 
    top_n: int=10
) -> list[dict]

基于提供的查询 query 和其他配置参数，在记忆中搜索并检索消息。

参数 (Parameters)

query: `string`, 必填

用于寻找相关消息的搜索词或自然语言问题。

memory_id: `list[str]`, 必填

要搜索的记忆 ID 列表。支持多个值。

agent_id: `string`, 可选

消息的来源智能体 ID。默认值为 None。

session_id: `string`, 可选

消息的会话 ID。默认值为 None。

user_id: `string`, 可选

参与智能体对话的用户 ID。默认值为 None。

similarity_threshold: `float`, 可选

消息被判定为匹配所需的最小余弦相似度分数。值越高，匹配结果越精准，但数量越少。默认值为 0.2。

范围 [0.0, 1.0]

keywords_similarity_weight: `float`, 可选

控制最终相关度评分中，关键词匹配与语义（基于嵌入向量）匹配的影响比例。值设为 0.5 时表示二者权重等同。默认值为 0.7。

范围 [0.0, 1.0]

top_n: `int`, 可选

返回的最相关消息的最大数量。这会限制结果集大小以提高效率。默认值为 10。

返回值 (Returns)

成功：表示消息字典的列表。

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.search_message("your question", ["your memory_id"])

获取近期消息 (Get Recent Messages)

RAGFlow.get_recent_messages(
    memory_id: list[str], 
    agent_id: str=None, 
    session_id: str=None, 
    limit: int=10
) -> list[dict]

从指定的记忆中检索最近的消息。通常接受 limit 参数以控制返回的消息数量。

参数 (Parameters)

memory_id: `list[str]`, 必填

要搜索的记忆 ID 列表。支持多个值。

agent_id: `string`, 可选

消息的来源智能体 ID。默认值为 None。

session_id: `string`, 可选

消息的会话 ID。默认值为 None。

limit: `int`, 可选

控制返回消息的数量。默认值为 10。

返回值 (Returns)

成功：表示消息字典的列表。

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
rag_object.get_recent_messages(["your memory_id"])

获取消息内容 (Get Message Content)

Memory.get_message_content(message_id: int)

使用其唯一的消息 ID，检索特定消息的完整内容和嵌入向量。

参数 (Parameters)

message_id: `int`, 必填

返回值 (Returns)

成功：表示消息的字典。

失败：Exception

示例 (Examples)

from ragflow_sdk import RAGFlow
rag_object = RAGFlow(api_key="<YOUR_API_KEY>", base_url="http://<YOUR_BASE_URL>:9380")
memory_object = Memory(rag_object, {"id": "your memory_id"})
memory_object.get_message_content(message_id)

错误码 (ERROR CODES)

兼容 OpenAI 的 API (OpenAI-Compatible API)

创建对话补全 (Create chat completion)

参数 (Parameters)

chat_id: string, 必填

model: string, 必填

messages: list[object], 必填

stream: boolean

返回值 (Returns)

示例 (Examples)

数据集管理 (DATASET MANAGEMENT)

创建数据集 (Create dataset)

参数 (Parameters)

name: string, 必填

avatar: string

description: string

permission

chunk_method, string

parser_config

返回值 (Returns)

示例 (Examples)

删除数据集 (Delete datasets)

参数 (Parameters)

ids: list[str] 或 None

delete_all: bool

返回值 (Returns)

示例 (Examples)

列出数据集 (List datasets)

参数 (Parameters)

page: int

page_size: int

orderby: string

desc: bool

id: string

name: string

include_parsing_status: bool

返回值 (Returns)

示例 (Examples)

列出所有数据集

根据 ID 检索数据集

列出包含解析状态的数据集

更新数据集 (Update dataset)

参数 (Parameters)

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

返回值 (Returns)

示例 (Examples)

数据集内文件管理 (FILE MANAGEMENT WITHIN DATASET)

上传文档 (Upload documents)

参数 (Parameters)

document_list: list[dict], 必填

返回值 (Returns)

示例 (Examples)

更新文档 (Update document)

参数 (Parameters)

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

返回值 (Returns)

示例 (Examples)

下载文档 (Download document)

返回值 (Returns)

示例 (Examples)

列出文档 (List documents)

参数 (Parameters)

id: string

keywords: string

page: int

page_size: int

orderby: string

desc: bool

create_time_from: int

create_time_to: int

返回值 (Returns)

示例 (Examples)

删除文档 (Delete documents)

参数 (Parameters)

ids: list[str] 或 None

delete_all: bool

返回值 (Returns)

示例 (Examples)

解析文档 (Parse documents)

参数 (Parameters)

chat_id: `string`, 必填

model: `string`, 必填

messages: `list[object]`, 必填

stream: `boolean`

name: `string`, 必填

avatar: `string`

description: `string`

chunk_method, `string`

ids: `list[str]` 或 `None`

delete_all: `bool`

page: `int`

page_size: `int`

orderby: `string`

desc: `bool`

id: `string`

name: `string`

include_parsing_status: `bool`

update_message: `dict[str, str|int]`, 必填

document_list: `list[dict]`, 必填

update_message: `dict[str, str|dict[]]`, 必填

id: `string`

keywords: `string`

page: `int`

page_size: `int`

orderby: `string`

desc: `bool`

create_time_from: `int`

create_time_to: `int`

ids: `list[str]` 或 `None`

delete_all: `bool`

document_ids: `list[str]`, 必填

document_ids: `list[str]`, 必填

document_ids: `list[str]`, 必填

content: `string`, 必填

important_keywords: `list[str]`

questions: `list[str]`

image_base64: `string`

tag_kwd: `list[str]`

keywords: `string`

page: `int`

page_size: `int`

id: `string`

ids: `list[str]` 或 `None`

delete_all: `bool`

update_message: `dict[str, str|list[str]|bool]` 必填

question: `string`, 必填

dataset_ids: `list[str]`, 必填

document_ids: `list[str]`

page: `int`

page_size: `int`

similarity_threshold: `float`

vector_similarity_weight: `float`

top_k: `int`

rerank_id: `string`

keyword: `bool`

cross_languages: `list[string]`

metadata_condition: `dict`

name: `string`, 必填

icon: `string`

dataset_ids: `list[str]`

llm_id: `str | None`

llm_setting: `dict | None`

prompt_config: `dict | None`