RAGFlow API 操作技能

概述

这个技能提供了与 RAGFlow 知识库系统交互的完整功能集，以及配套的数据库管理功能。RAGFlow 是一个开源的 RAG (检索增强生成) 知识库系统，用于管理文档和进行智能检索。

核心功能：

• RAGFlow 操作：知识库管理、文档管理、知识检索、RAG 对话
• 数据库交互：知识库文档管理、研究任务管理、研究报告管理

快速开始

1. 配置环境变量

在使用前，请设置以下环境变量：

RAGFlow 配置

# Windows PowerShell
$env:RAGFLOW_BASE_URL = "http://your-ragflow-server:9380"
$env:RAGFLOW_API_KEY = "your-api-key"

# Linux/Mac
export RAGFLOW_BASE_URL="http://your-ragflow-server:9380"
export RAGFLOW_API_KEY="your-api-key"

数据库配置（SQLite - 默认）

# 可选：指定 SQLite 数据库文件路径
$env:SQLITE_PATH = "./ragflow_data.db"

数据库配置（MySQL）

# Windows PowerShell
$env:DB_TYPE = "mysql"
$env:DB_HOST = "localhost"
$env:DB_PORT = "3306"
$env:DB_USER = "root"
$env:DB_PASSWORD = "your-password"
$env:DB_NAME = "ragflow_db"

# Linux/Mac
export DB_TYPE="mysql"
export DB_HOST="localhost"
export DB_PORT="3306"
export DB_USER="root"
export DB_PASSWORD="your-password"
export DB_NAME="ragflow_db"

2. 基本使用示例

RAGFlow 客户端

from scripts.ragflow_client import RagflowClient

# 初始化客户端
client = RagflowClient()

# 检查服务是否可用
if client.available:
    print("RAGFlow 服务可用")
    
    # 列出所有知识库
    kbs = client.get_knowledge_bases()
    for kb in kbs:
        print(f"知识库: {kb.get('name')} (ID: {kb.get('id')})")

数据库客户端

from scripts.db_client import DatabaseClient

# 初始化数据库客户端
db = DatabaseClient()

# 初始化数据库（创建表结构）
if db.initialize():
    print("数据库初始化成功")
    
    # 创建知识库文档记录
    doc = db.create_knowledge_document(
        title="示例文档",
        source_type="PDF",
        ragflow_dataset_id="kb-001",
        ragflow_document_id="doc-001"
    )
    print(f"文档记录创建成功，ID: {doc.id}")

核心功能

RAGFlow 操作

知识库管理

列出所有知识库

kbs = client.get_knowledge_bases()
for kb in kbs:
    print(f"名称: {kb.get('name')}")
    print(f"ID: {kb.get('id')}")
    print(f"描述: {kb.get('description', '')}")
    print(f"文档数: {kb.get('document_count', 0)}")

创建知识库

# 创建知识库
kb = client.create_knowledge_base(
    name="我的知识库",
    description="这是一个示例知识库"
)

if kb:
    print(f"知识库创建成功，ID: {kb.get('id')}")

删除知识库

success = client.delete_knowledge_base("knowledge-base-id")
if success:
    print("知识库删除成功")

文档管理

列出知识库中的文档

docs = client.get_documents("knowledge-base-id")
for doc in docs:
    print(f"文档: {doc.get('name')}")
    print(f"ID: {doc.get('id')}")
    print(f"状态: {doc.get('status')}")
    print(f"片段数: {doc.get('chunk_count', 0)}")

上传文档

# 上传文档到知识库
doc = client.upload_document(
    kb_id="knowledge-base-id",
    file_path="path/to/your/document.pdf"
)

if doc:
    print(f"文档上传成功，ID: {doc.get('id')}")

支持的文档格式：

• PDF (.pdf)
• Word (.docx, .doc)
• PowerPoint (.pptx, .ppt)
• Excel (.xlsx, .xls)
• 文本文件 (.txt, .md, .json, .csv)
• 图片 (.jpg, .png, .jpeg)

删除文档

success = client.delete_document(
    kb_id="knowledge-base-id",
    doc_id="document-id"
)
if success:
    print("文档删除成功")

知识检索

简单检索

# 在知识库中检索相关内容
results = client.retrieve(
    kb_id="knowledge-base-id",
    query="什么是人工智能？",
    top_k=5
)

print(f"找到 {len(results)} 个相关结果")
for i, result in enumerate(results, 1):
    print(f"\n--- 结果 {i} (分数: {result.get('score', 0):.3f}) ---")
    print(result.get('content', ''))

RAG 对话

简单 RAG 对话

simple_chat 方法会检索知识库并返回格式化的上下文和提示词，方便你结合 LLM 使用：

# 进行 RAG 对话
chat_result = client.simple_chat(
    kb_id="knowledge-base-id",
    question="请介绍一下这个项目的主要功能",
    top_k=5
)

# 输出来源
print("参考来源:")
for source in chat_result.get('sources', []):
    print(f"  - {source}")

# 使用返回的提示词与 LLM 对话
print("\n系统提示词:")
print(chat_result.get('system_prompt', ''))
print("\n用户提示词:")
print(chat_result.get('user_prompt', ''))

数据库交互

初始化数据库

from scripts.db_client import DatabaseClient

# 初始化数据库客户端
db = DatabaseClient()

# 初始化数据库（创建表结构）
if db.initialize():
    print("数据库初始化成功")

知识库文档管理

# 创建知识库文档记录
doc = db.create_knowledge_document(
    title="示例文档",
    source_type="PDF",
    source_url="https://example.com/doc.pdf",
    content_summary="文档内容摘要",
    metadata_info={"author": "张三", "pages": 10},
    ragflow_dataset_id="kb-001",
    ragflow_document_id="doc-001"
)
print(f"文档记录创建成功，ID: {doc.id}")

# 列出知识库文档
docs, total = db.list_knowledge_documents(
    source_type="PDF",
    search="示例",
    skip=0,
    limit=10
)
print(f"找到 {total} 个文档")
for doc in docs:
    print(f"  - {doc.title} (ID: {doc.id})")

# 更新文档
updated = db.update_knowledge_document(
    doc_id=1,
    title="更新后的标题",
    is_parsed=True
)

# 删除文档
success = db.delete_knowledge_document(doc_id=1)

从 RAGFlow 同步文档

# 从 RAGFlow 文档数据同步到数据库
ragflow_doc = {
    "id": "doc-001",
    "name": "技术文档.pdf",
    "run": "SUCCESS"
}
synced_doc = db.sync_from_ragflow_document(
    ragflow_doc=ragflow_doc,
    kb_id="kb-001",
    source_type="技术文档"
)

研究任务管理

# 创建研究任务
task = db.create_research_task(
    company_name="示例公司",
    research_title="市场分析报告",
    research_goal="分析市场趋势和竞争对手",
    focus_domains=["市场", "竞争"],
    output_depth="detailed",
    total_iterations=5
)
print(f"任务创建成功，ID: {task.id}")

# 列出研究任务
tasks, total = db.list_research_tasks(
    company_name="示例公司",
    status="pending"
)

# 更新任务状态
updated = db.update_research_task_status(
    task_id=1,
    status="in_progress",
    progress=50,
    current_step="数据收集"
)

研究报告管理

# 创建研究报告
report = db.create_research_report(
    title="2024年市场分析报告",
    company_name="示例公司",
    research_goal="分析市场趋势",
    file_path="/reports/market_2024.md",
    status="draft"
)

# 列出研究报告
reports, total = db.list_research_reports(
    company_name="示例公司"
)

命令行使用

ragflow_client.py 也可以作为命令行工具使用：

# 列出所有知识库
python scripts/ragflow_client.py --list-kbs

# 列出指定知识库的文档
python scripts/ragflow_client.py --list-docs <kb-id>

# 创建知识库
python scripts/ragflow_client.py --create-kb "知识库名称" "知识库描述"

# 上传文档
python scripts/ragflow_client.py --upload <kb-id> "path/to/document.pdf"

# 检索知识
python scripts/ragflow_client.py --retrieve <kb-id> "查询文本" 5

资源

scripts/ragflow_client.py

完整的 RAGFlow API 客户端实现，包含所有核心功能。请查看此文件了解详细的 API 使用方法。

scripts/db_client.py

数据库交互客户端，提供知识库文档管理、研究任务管理、研究报告管理等功能。支持 SQLite 和 MySQL 数据库。

references/api_reference.md

RAGFlow API 的详细参考文档，包括端点说明、请求/响应格式和错误处理。

最佳实践

1. 错误处理：始终检查 client.available 确保服务可用
2. 异常捕获：API 调用可能失败，建议使用 try-except 包裹
3. 超时设置：上传大文件时可能需要更长时间，请耐心等待
4. API 密钥安全：不要将 API 密钥提交到代码仓库，使用环境变量
5. 文档命名：使用有意义的文档名称，便于后续管理和检索

故障排除

连接失败

• 检查 RAGFLOW_BASE_URL 是否正确
• 确认 RAGFlow 服务正在运行
• 检查网络连接和防火墙设置

认证错误

• 确认 RAGFLOW_API_KEY 已正确设置
• 检查 API 密钥是否有效且未过期

上传失败

• 确认文档格式是否受支持
• 检查文件大小是否超出限制
• 确认知识库 ID 是否正确