使用Strands SDK和RAGFlow构建具备RAG能力Agent理应用

zhaojiew10

375人浏览 · 2026-03-12 12:58:05

zhaojiew10 · 2026-03-12 12:58:05 发布

在企业数字化转型的浪潮中，如何让 AI 助手能够准确回答公司内部的专业问题，一直是个挑战。传统的大语言模型虽然知识丰富，但对企业特有的规章制度、技术文档却一无所知。RAG（检索增强生成）技术的出现，为这个问题提供了优雅的解决方案。本文使用开源的 RAGFlow 和 Strands Agent 框架，创建一个简单的具备RAG能力的智能助手。

RAGFlow基础概念

在众多 RAG 解决方案中，RAGFlow 脱颖而出的原因：

开箱即用的文档解析能力：支持 PDF、Word、Markdown、Excel 等多种格式，无需手动处理文档格式转换。
深度文档理解：不是简单的文本切分，而是理解文档结构、表格、图片，提取更精准的知识块。
灵活的向量检索：支持多种 embedding 模型，可以根据场景选择最合适的模型。
完善的 Python SDK：相比直接调用 HTTP API，SDK 提供了更友好的接口和错误处理。

我们的系统采用三层架构：

知识层（RAGFlow）：负责文档存储、解析和向量检索。这一层将非结构化的文档转换为可检索的知识块。
工具层（search_knowledge）：封装 RAGFlow 的检索能力，为 Agent 提供标准化的工具接口。
智能层（Strands Agent）：理解用户意图，决定何时调用检索工具，并基于检索结果生成自然语言回答。

RAGFlow 如何组织知识？主要有

Dataset

Dataset（数据集）是 RAGFlow 中的顶层概念，类似于传统数据库中的 Database。每个 Dataset 代表一个独立的知识领域，比如"HR 政策"、“技术文档”、“客服知识库”。为什么需要 Dataset？因为不同领域的知识可能需要不同的处理策略。HR 文档可能更注重精确匹配，而技术文档可能需要更宽松的语义理解。每个 Dataset 可以配置独立的 embedding 模型和解析策略。

Document

Document 是具体的文件，比如一份员工手册、一个技术规范。上传到 RAGFlow 后，Document 会经历解析过程，这个过程包括识别文档结构（标题、段落、表格），提取文本内容，将内容切分成合适大小的块，为每个块生成向量表示

解析是异步的，这意味着上传后不会立即可用，需要等待几秒到几分钟（取决于文档大小）。

Chunk

Chunk 是文档解析后的知识块，通常是几百字的文本片段。为什么要切分成 Chunk？因为：

检索精度：小块更容易匹配用户的具体问题
上下文限制：LLM 的输入长度有限，不能把整个文档都塞进去
相关性排序：可以返回最相关的几个片段，而不是整个文档

一个好的 Chunk 应该是语义完整的，比如一个完整的段落或一个小节。RAGFlow 的智能解析会尽量保持语义完整性。

此外，最开始的时候使用了原始的 HTTP 请求来调用 RAGFlow API。代码看起来是这样的：

response = httpx.post(
    f"{RAGFLOW_BASE_URL}/retrieval",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={"question": "...", "dataset_ids": [...]}
)
result = response.json()
chunks = result.get("data", {}).get("chunks", [])

这种方式能工作，但有几个问题，代码可读性差：充斥着 URL 拼接、JSON 构造，业务逻辑被淹没在技术细节中。后来发现 RAGFlow 提供了官方 Python SDK，SDK 不仅简化了代码，还提供了类型提示、自动重试、连接池管理等企业级特性。

from ragflow_sdk import RAGFlow

rag = RAGFlow(api_key=API_KEY, base_url=RAGFLOW_URL)
chunks = rag.retrieve(dataset_ids=[DATASET_ID], question="...")

敏感信息（API Key、数据库连接）不应该硬编码在代码中。我们使用 .env 文件管理配置：

OPENAI_API_KEY=sk-xxx
OPENAI_BASE_URL=http://your-llm-endpoint/v1
RAGFLOW_URL=http://localhost:9380
RAGFLOW_API_KEY=ragflow-xxx
DATASET_ID=your-dataset-id

核心功能实现

文档上传

文档上传是整个系统的入口，但实现起来有些细节需要注意。RAGFlow SDK 的 upload_documents() 方法要求特定的数据格式：

docs = dataset.upload_documents([{
    "display_name": "员工手册.md",
    "blob": file_object
}])

注意这里是一个字典列表，而不是直接传文件路径。blob 是文件对象，需要用 open() 打开。这种设计允许批量上传，提高效率。以下文件上传成功，但是还没有触发解析。

其次，上传后需要触发解析。SDK 提供了两个方法：

parse_documents()：同步方法，会阻塞直到解析完成
async_parse_documents()：异步方法，立即返回

在生产环境中，应该使用异步方法，避免长时间阻塞。上传脚本可以立即返回，让用户继续其他操作，解析在后台进行。

状态检查

文档上传后，如何知道解析是否完成？这需要轮询文档状态。

每个 Document 对象有两个关键属性：

run：解析状态，可能是 DONE、FAIL、CANCEL 或处理中
chunk_count：解析出的 Chunk 数量

一个简单的检查逻辑：

for doc in dataset.list_documents():
    if doc.run == 'DONE' and doc.chunk_count > 0:
        print(f"✓ {doc.name} 已就绪")
    else:
        print(f"⏳ {doc.name} 处理中...")

在实际应用中，可以实现一个后台任务，定期检查文档状态，并在解析完成后发送通知。

知识检索

检索是 RAG 系统的灵魂。RAGFlow 的 retrieve() 方法看似简单，但背后是复杂的向量检索和排序算法。

关键参数解析：

top_k：返回最相关的 K 个结果。设置太小可能遗漏重要信息，设置太大会引入噪音。经验值是 3-10，具体取决于文档密度。
similarity_threshold：相似度阈值，低于此值的结果会被过滤。这是质量和召回率的权衡。设置 0.1 比较宽松，0.5 比较严格。
page_size：分页大小，与 top_k 类似但用于分页场景。

一个常见的误区是认为 top_k 越大越好。实际上，过多的检索结果会稀释 LLM 的注意力，反而降低回答质量。少而精胜过多而杂。

Agent 集成

Strands Agent 框架的核心思想是"工具使用"。我们不是直接把检索结果塞给 LLM，而是让 LLM 自己决定何时需要检索。

首先需要明确的工具定义，使用 @tool 装饰器定义工具，包含清晰的文档字符串，说明工具的功能、参数和返回值。

@tool
def search_knowledge(question: str, top_k: int = 5) -> str:
    """Search RAGFlow knowledge base for relevant information.
    
    Args:
        question: The question to search for
        top_k: Number of top results to return
    
    Returns:
        Retrieved knowledge chunks as formatted text
    """
    # 实现检索逻辑

其次需要有明确的 System Prompt告诉 Agent 何时应该使用工具。注意这里用了 “MUST”，强调必须调用工具。如果 Prompt 不够明确，Agent 可能会凭借自己的知识回答，而不去检索，导致答案不准确。

system_prompt=(
    "You are a helpful assistant. "
    "When users ask questions, you MUST use the search_knowledge tool "
    "to search the knowledge base first before answering."
)

完整示例

接下来通过一个完整的案例，看看系统如何工作。例如，公司 HR 部门有一份详细的员工手册，包含考勤制度、薪酬福利、培训发展等内容。新员工经常询问年假政策、福利待遇等问题，HR 需要反复解答。我们希望构建一个 AI 助手，能够自动回答这些常见问题。

准备知识库

首先创建员工手册文档，使用 Markdown 格式便于解析，文档使用标题层级组织内容，这样 RAGFlow 可以更好地理解文档结构。

# XX科技有限公司员工手册

## 第二章 考勤制度

### 2.2 请假制度
- 年假：
  - 入职满1年：5天
  - 入职满3年：7天
  - 入职满5年：10天
  - 入职满10年：15天

运行上传脚本：

uv run python document_uploader.py hr_docs/员工手册.md

几秒钟后，文档上传完成并开始解析。这时可以通过状态检查脚本查看进度：

uv run python check_and_parse.py

输出显示文档已解析成 5 个 Chunks，说明系统将员工手册分成了 5 个语义完整的片段。

此外，由于我们设置了自动元数据补充，还能够自动加入元数据字段增加后续检索的精度

测试检索

在让 Agent 使用之前，先单独测试检索功能：

uv run python test_retrieval.py

系统返回了 3 个相关的 Chunks，相似度分别是 0.566、0.560、0.550。第一个 Chunk 正好包含年假政策的内容，说明检索工作正常。

现在让 Agent 回答问题：

uv run python ragflow_agent.py

观察输出，可以看到 Agent 的工作流程：

识别到用户询问年假政策
调用 search_knowledge 工具
工具返回检索到的知识片段
Agent 基于检索结果，用自然语言生成回答

最终回答准确地列出了不同司龄对应的年假天数，还补充了未休年假的补偿政策。这些信息都来自知识库，而不是 LLM 的预训练知识。

扩展方向

这里只是简单的实现一个示例，来展示ragflow的rag能力如何集成到agent中，后续可以考虑如下方式扩展

多数据集联合检索

当企业有多个知识库时，可以同时检索多个 Dataset：

chunks = rag.retrieve(
    dataset_ids=[HR_DATASET_ID, TECH_DATASET_ID],
    question=question
)

这样 Agent 可以综合多个领域的知识回答问题。但要注意，多数据集检索会增加延迟和成本。

混合检索策略

RAGFlow 支持向量检索和关键词检索的混合：

chunks = rag.retrieve(
    dataset_ids=[DATASET_ID],
    question=question,
    keyword=True,
    vector_similarity_weight=0.7
)

向量检索擅长语义理解，关键词检索擅长精确匹配。混合使用可以兼顾两者优势。

元数据过滤

为文档添加元数据（如部门、类型、日期），检索时可以过滤：

chunks = rag.retrieve(
    dataset_ids=[DATASET_ID],
    question=question,
    metadata_condition={"department": "HR", "year": "2024"}
)

这在大型知识库中特别有用，可以缩小检索范围，提高精度。

亚马逊云科技技术品牌专区

更多推荐

Hugging Face Optimum 源码静态评测：模型部署优化框架的工程结构与验证边界

亚马逊云科技技术品牌专区

智能软件工程AI4SE（一）——智能需求工程

亚马逊云科技技术品牌专区

车载OBD的6路摄像头同时传，画面卡？选对物联网卡立马流畅回传

亚马逊云科技技术品牌专区

所有评论(0)

查看更多评论

zhaojiew10

@sinat_41567654

已为社区贡献3条内容

使用Strands SDK和RAGFlow构建具备RAG能力Agent理应用

zhaojiew10

RAGFlow基础概念

Dataset

Document

Chunk

核心功能实现

文档上传

状态检查

知识检索

Agent 集成

完整示例

准备知识库

测试检索

扩展方向

多数据集联合检索

混合检索策略

元数据过滤

所有评论(0)

温馨提示：您尚未绑定手机号

zhaojiew10