最近在推进一个智能问答项目做的事情正好就是标题里这句用 Strands SDK 搭配 RAGFlow把企业知识库变成 Agent 的实时记忆。先说结论这套组合确实解决了之前靠文本切块向量检索硬怼时最头疼的几个问题比如表格解析错位、长文档召回碎片化以及 Agent 在决定要不要检索、检索引擎怎么被调用时的盲目和失控。这篇文章会按一条可以复现的路径来讲为什么拆成知识管线和 Agent 编排两部分、怎么部署初始化、索引和检索参数怎么调、Strands SDK 怎么把 RAGFlow 封装成 Agent 的工具最后是真实场景下的压测数据和四个典型踩坑。适合正在做 Agent 问答、企业内部知识库增强检索的开发者参考。1. 为什么拆成知识管线和Agent 编排两半1.1 普通 RAG 与 Agent RAG 的本质差别普通 RAG 的链路是一条直路用户问题 → 向量化 → 检索 TopK → 拼接上下文 → 模型回答。这个模式对单一知识库、固定问答场景够用但你一旦把知识库放开到几十个文档、几百个章节问题就开始变形了。用户问我们退货政策里退货周期是几天普通 RAG 检索一次就能答。但如果问题变成对比 A 产品和 B 产品的售后条款差异你需要查两次库、分别取回 A 和 B 的内容再把两组结果交给模型做比较。更进一步帮我根据 R100 的保修条款给客户写一封解释延保政策的邮件这已经不是检索一次拼答案能覆盖的而是模型需要先规划我要查保修条款 → 查完读内容 → 组织成邮件。这正是 Agent 循环在做的事。所以普通 RAG 是把检索结果当成提示词里的一段上下文而 Agent RAG 是把检索结果当成一个工具函数的返回值。区别看似不大工程落地上完全是两套设计前者是数据管道后者是决策框架。1.2 RAGFlow 扛下文档理解这段脏活自己搭 RAG 最坑的不是向量库而是文档进库之前那道解析关。企业知识库里 PDF 居多里面有表格、图片、页眉页脚、扫描件甚至多层嵌套表格。传统做法是直接把 PDF 文本抽出来按固定长度切块结果表格被切得七零八落检索时明明库里有关键信息却因为分块太碎召不回来。RAGFlow 的价值在于把文档理解做成了完整管线版面分析、表格结构还原、OCR、分块策略、向量化和混合检索。它不是简单地把文本按字符数切开而是会尝试理解页面布局把表格当表格处理把关联段落归并到一起。这个能力对中文企业文档尤其关键因为表格往往是产品参数和售后条款的集中地。我用它做了个简单对比同一份 50 页产品手册传统切块方式对R100 的整机保修期是多久这个问题召回的命中段落是第 4 块实际答案在第 42 页表格里而 RAGFlow 解析后能正确把第 42 页的表格针对保修期字段单独切出来。差距就在这一步体现。1.3 Strands SDK 负责的另一半决策和执行RAGFlow 只回答检索结果是什么它不负责 Agent 要不要查、查完怎么用。这部分由 Strands SDK 接管。Strands SDK 是一个云原生的 Agent 编排 SDK核心能力是三块工具注册、Agent 主循环推理、调用、观察、再推理、会话上下文管理。它自己不做文档解析也不做检索但它能把RAGFlow 的检索 API包装成一个普通工具函数注册给 Agent。模型在对话中判断这个问题需要查知识库就会生成一次函数调用Strands SDK 负责调用我们的工具把 RAGFlow 返回的检索片段回传给模型模型再基于这些片段组织回答。概括一句话RAGFlow 负责让知识进得来、找得到Strands SDK 负责让 Agent想得到、用得对。两者之间通过 HTTP API 连接耦合度非常低这也是我选这套架构的核心理由。2. 部署与初始化从 Docker 到第一份文档入库2.1 RAGFlow 部署时最容易被低估的资源配置RAGFlow 的部署表面很简单Docker Compose 一条命令就能把服务拉起来默认端口 9380。真正的坑在资源规划上。我第一次只给容器分配了 8G 内存然后一次性上传了 200 多页的混合文档结果服务在解析任务跑到一半时直接假死页面打不开接口全部超时。后来把内存加到 16G同时在配置里限制了同时解析的文档数量我这边限制为 2 个整套系统才稳定下来。磁盘也需要预留空间。除了原始文档RAGFlow 还会保存解析后的中间产物以及向量索引一份 5M 的 PDF 解析完索引可能膨胀到 50~100M。长期跑几十个数据集磁盘增长很快建议至少预留 50G 以上。services: ragflow-server: image: ragflow:your-version ports: - 9380:80 environment: - SVR_HTTP_PORT9380 volumes: - ./datasets:/ragflow/datasets mem_limit: 16g2.2 初始化 Strands SDK 环境Strands SDK 用 Python 包分发建议在 Python 3.10 以上的环境里使用。安装很简单pip install strands-sdk初始化时要注意Strands SDK 和 RAGFlow 是两套独立的凭据体系。Strands SDK 面向大模型对话服务需要配置模型网关地址、API Key、默认模型RAGFlow 则是自己的 API Key。两者不要混用不然排查问题时会浪费很多时间。from strands_sdk import Agent agent Agent( modelos.getenv(AGENT_MODEL, your-llm-endpoint), api_keyos.getenv(AGENT_API_KEY), )提示如果你选的是别的 Agent 编排 SDK只需要保留工具注册和主循环的概念后半部分代码思路同样适用。2.3 创建数据集并完成第一次入库RAGFlow 里组织知识的最小单位是数据集Dataset。每个数据集对应一个独立的知识库可以配置单独的解析策略和检索范围。完整流程是登录控制台创建数据集 → 通过 API 上传文档 → 触发解析 → 轮询解析状态。这里我直接给一段可跑的 Python 脚本后面章节还会复用它。import os import time import requests RAGFLOW_URL os.getenv(RAGFLOW_URL, http://localhost:9380) RAGFLOW_API_KEY os.getenv(RAGFLOW_API_KEY, ragflow-xxxx) HEADERS {Authorization: fBearer {RAGFLOW_API_KEY}} def create_dataset(name: str) - str: resp requests.post( f{RAGFLOW_URL}/api/v1/datasets, headersHEADERS, json{name: name}, ) resp.raise_for_status() return resp.json()[data][id] def upload_document(dataset_id: str, file_path: str) - str: with open(file_path, rb) as f: resp requests.post( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents, headersHEADERS, files{file: f}, ) resp.raise_for_status() return resp.json()[data][0][id] def parse_document(dataset_id: str, document_id: str) - None: resp requests.post( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents/{document_id}/chunks, headersHEADERS, ) resp.raise_for_status() def wait_until_ready(dataset_id: str, document_id: str, timeout: int 600) - str: start time.time() while time.time() - start timeout: resp requests.get( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents/{document_id}, headersHEADERS, ) status resp.json()[data][run] if status DONE: return status if status in (FAIL, CANCEL): raise RuntimeError(fparse failed: {status}) time.sleep(5) raise TimeoutError(parse timeout) dataset_id create_dataset(product-manual) doc_id upload_document(dataset_id, ./docs/产品手册.pdf) parse_document(dataset_id, doc_id) print(wait_until_ready(dataset_id, doc_id))第一次跑通这个流程你会看到文档状态从待解析变成解析中最后变成已解析或已切分。这一步成功之后后续的检索调优才有意义。3. 索引与检索调优决定 RAG 质量的前置工作3.1 解析模板不是越复杂越好RAGFlow 提供多种解析模板常见的大类有通用、问答、表格、手动。很多人上来一律用通用遇到表格多的文档就吃亏。我整理了一个选型参考基于同一批 20 份不同结构文档的实测效果文档类型推荐模板原因我的补充排版规整的 Word/Markdown通用段落语义相对完整切分稳定注意保留标题层级含大量表格的参数手册表格优先表格会被还原成结构化片段配合小 chunk 使用FAQ 或一问一答式文档问答按问题-答案粒度切分召回命中率高对检索问题非常友好扫描件/图片型 PDF通用 OCR 开启先识别文字再做版面分析识别质量取决于 PDF 清晰度不要迷信自动或最复杂的模板。解析模板的本质是决定 chunk 的边界在哪里选错边界后面无论怎么调检索参数都救不回来。3.2 chunk 大小和重叠度的设定方法chunk 大小是我调 RAG 时最反复的一个参数。太小语义碎片化检索召回的片段常常只有半句话模型拿不到来龙去脉太大TopK 检索拼出来的上下文又长又杂模型回答容易跑偏。我目前的基线是通用文档 chunk 大小 512 个 token重叠 80 个 token。这个数字不是拍脑袋来的——我测过 256、512、1024 三档512 在召回准确率和上下文信噪比之间最平衡。重叠度是为了避免恰好把一个完整段落切在中间导致语义断裂。80 个 token 大约等于 4~6 行中文文本足够把上文的指代关系缓冲住。表格类文档我会把 chunk 缩小到 256 甚至更小因为表格片段本身就是自包含的切大反而容易把不相干的列混进来。3.3 检索参数TopK、相似度阈值和混合权重RAGFlow 的检索接口有三个参数最值得关注top_k召回数量。Agent 场景我推荐 5~8。太少答案容易漏太多上下文塞满噪音模型反而分不清主次。similarity_threshold相似度阈值低于该分数的结果会被丢弃。推荐从 0.2 开始调不要一上来设 0.5 以上否则很多语义相关但字面不同的表达会被过滤掉。vector_similarity_weight混合检索中向量相似度的权重剩余权重给全文检索。这个参数很多人不理解简单说就是0.1~0.3 偏全文关键词适合产品型号、代码片段这类精确匹配场景0.7~0.9 偏语义适合自然语言问答、口语化提问。我的经验是先按向量权重 0.5、阈值 0.2、top_k 5跑一轮再看失败案例逐个调整。不要上来就追求完美参数因为没有一套参数能同时适配所有文档。3.4 用测试集验证召回而不是凭感觉调参调参最大的敌人是感觉。我建议建一个标准测试集从知识库里挑 20~30 个典型问题人工标注每个问题的答案出自哪份文档的哪个段落或者至少标注出自哪一分部内容。然后写脚本批量调 RAGFlow 的检索接口统计召回命中率。我自己用的评测表格长这样问题期望来源召回命中top_k3top_k5top_k8R100 整机保修期多久产品手册-保修表格是否是是退货需要提供哪些凭证售后政策-第 3 章是是是是设备 IP 地址怎么改快速上手-网络配置是否否是不要偷懒跳过这步。没有测试集你对参数的所有调整都只是看起来更好了而不是确实更好了。4. 用 Strands SDK 把检索变成 Agent 的可用工具4.1 工具注册机制Strands SDK 的工具注册方式很直接一个普通 Python 函数加上装饰器函数签名会自动变成模型可理解的工具描述函数的 docstring 会被当作工具的使用说明。模型看不到函数内部实现它只看到这个工具叫什么、有什么参数、大概能干什么。所以函数名和参数名要起得让模型一看就懂。比如query_knowledge_base(question: str, top_k: int 5)就比query_kb(q: str, k: int 5)友好得多。docstring 里要写清楚工具的边界在企业知识库中检索与问题最相关的段落让模型知道这个工具适合处理什么类型的问题。4.2 封装一个 query_knowledge_base 工具直接上代码这个工具的核心是把 RAGFlow 的/api/v1/retrieval接口封装成 Strands 的工具函数from strands_sdk import tool DATASET_ID your-dataset-id tool def query_knowledge_base(question: str, top_k: int 5) - str: 在企业知识库中检索与问题最相关的段落返回内容会标注来源文档和段落编号。 当用户问题涉及产品参数、操作流程、售后政策、内部规范时必须先调用本工具。 payload { question: question, dataset_ids: [DATASET_ID], top_k: top_k, similarity_threshold: 0.2, vector_similarity_weight: 0.3, } resp requests.post( f{RAGFLOW_URL}/api/v1/retrieval, headersHEADERS, jsonpayload, timeout30, ) resp.raise_for_status() chunks resp.json()[data][chunks] if not chunks: return 知识库中没有找到相关内容。 lines [f[{c[doc_name]}#{c[chunk_id]}] {c[content]} for c in chunks] return \n.join(lines)注意返回格式里每一行都带[文档名#段落号]这是刻意设计的。后面模型回答时必须引用这些编号否则工具返回值里的来源信息就被浪费了。4.3 在 System Prompt 里写清工具的调用边界Agent 不会无缘无故调用工具它全靠系统提示词来决定行动。我踩过的最大的坑是没有告诉模型什么时候不要检索结果模型啥问题都先检索一遍回答慢、消耗高还经常被不相关段落带偏。现在的做法是在系统提示词里写清楚触发条件和跳过条件你是企业内部知识助手。 当用户问题涉及产品参数、操作流程、售后政策、内部规范时 必须先调用 query_knowledge_base 获取信息再基于检索结果回答。 回答中必须使用 [来源:文档名#段落号] 标注依据。 如果检索结果中没有答案直接说明未找到不要编造。 当用户只是闲聊、寒暄或者问题明显与知识库无关时不要调用检索工具。这段提示词看着简单实际调试时每一条都对应过具体的失败案例。比如不要编造这一条就是为了防止模型基于检索片段里的一句话强行发挥出根本不存在的结论。4.4 带引用溯源的回答是降低幻觉的关键RAG Agent 最危险的不是答错而是答得像真的一样。引用溯源是我目前能想到的最有效的一道防线要求模型在回答中显式标注依据来源。工程上的实现就是在检索工具返回值里把doc_name和chunk_id一起传回给模型并在提示词里强制要求引用。模型见过没有来源就直说没找到的约束之后编造率会明显下降。前端展示上我也做了配套每个回答下方列出引用卡片用户点一下就能跳到原始文档位置或者至少在界面上展示来源文档名和段落号。这一步对内部知识工具来说几乎是刚需因为用户需要验证答案的可信度。5. 真实场景实测压测数据与四个典型踩坑5.1 用 50 页产品手册做的压测我用一份 50 页的中文产品手册含 3 个大型参数表格、20 张界面截图、10 个 FAQ 段落做了压测。测试集 20 个问题对比不同参数组合参数组合召回命中率单次检索平均耗时模型回答可用率备注top_k3, v_weight0.5, 阈值0.370%0.8s60%多数表格类问题漏召回top_k5, v_weight0.3, 阈值0.290%1.2s85%当前基线top_k8, v_weight0.7, 阈值0.195%1.8s75%召回高但上下文噪音大结论很明确召回率高不等于回答可用率高。top_k8 那组虽然把答案都找回来了但模型被太多无关片段干扰回答反而变得含糊。最终我停在 top_k5 附近。5.2 踩坑一批量解析任务把服务拖死这是部署阶段遇到的。一次性上传 30 份文档并全部触发解析RAGFlow 服务在几分钟后失去响应。原因很直接所有文档同时做版面分析和 OCR内存和 CPU 直接被打满。解决方式是在应用层加了一个解析队列每次只触发 1~2 份文档的解析解析完成后再触发下一批。虽然整体变慢但服务不再假死。如果你的团队没有队列基础设施写一个简单的脚本循环也能解决。5.3 踩坑二检索接口同步超时引发连锁反应Agent 调用query_knowledge_base时RAGFlow 检索接口偶尔会超过 30 秒才返回集中在数据集特别大、文档 chun 数量多的场景。Strands SDK 的工具调用如果超时整个 Agent 对话会被卡住体验非常差。我做了三层防护工具函数设置 HTTP 超时30 秒超时后返回知识库暂时不可用请稍后重试而不是直接抛出异常同时给数据集做了拆分一个超大知识库拆成多个子数据集缩减单次检索范围。拆库之后检索耗时基本稳定在 2 秒以内。5.4 踩坑三相似度阈值设太高导致空召回有阵子我为了提高回答精确度把similarity_threshold提到 0.6。结果用户用口语化表达提问时召回结果常常为空Agent 回复未找到相关内容。原因很简单口语表达和文档原文的字面差异很大向量相似度分数本来就偏低阈值一高全被过滤了。阈值设定要结合你的检索类型来考虑。如果用户提问偏口语阈值 0.1~0.2 比较稳如果知识库是代码、规范、型号这种精确匹配场景阈值提高有意义同时要把向量权重调低、全文检索权重调高。5.5 踩坑四历史会话里的旧检索污染下一轮回答这个坑很隐蔽。Agent 的多轮对话中上一轮的检索结果会留在上下文里。如果用户接下来问了一个新话题模型的注意力仍可能被旧检索片段牵走回答新问题时混入旧文档的内容。解决方式Strands SDK 支持在每次工具调用后对旧的工具输出做截断或摘要我配置成只保留当前轮次的完整检索结果上一轮的工具输出压缩成一句摘要。这个配置之后多轮对话的串味问题基本消失。5.6 我目前使用的参数基线放在这里直接参考配置项推荐值适用场景chunk size / overlap512 / 80通用文档表格类 chunk256 / 32含大量表格的知识库top_k5Agent 对话检索similarity_threshold0.2口语化提问vector_similarity_weight0.3精确匹配为主Agent 工具超时30sHTTP 调用解析并发数2自部署低配服务器6. 可以直接落地的完整示例6.1 知识库索引脚本合并把前面分散的代码合并成一份可执行的脚本直接保存运行即可完成建库、传文档、解析、等待就绪全流程# index_knowledge_base.py import os import time import requests RAGFLOW_URL os.getenv(RAGFLOW_URL, http://localhost:9380) RAGFLOW_API_KEY os.getenv(RAGFLOW_API_KEY, ragflow-xxxx) HEADERS {Authorization: fBearer {RAGFLOW_API_KEY}} def create_dataset(name: str) - str: resp requests.post(f{RAGFLOW_URL}/api/v1/datasets, headersHEADERS, json{name: name}) resp.raise_for_status() return resp.json()[data][id] def upload_document(dataset_id: str, file_path: str) - str: with open(file_path, rb) as f: resp requests.post( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents, headersHEADERS, files{file: f}) resp.raise_for_status() return resp.json()[data][0][id] def parse_document(dataset_id: str, document_id: str) - None: resp requests.post( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents/{document_id}/chunks, headersHEADERS) resp.raise_for_status() def wait_until_ready(dataset_id: str, document_id: str, timeout: int 600) - str: start time.time() while time.time() - start timeout: resp requests.get( f{RAGFLOW_URL}/api/v1/datasets/{dataset_id}/documents/{document_id}, headersHEADERS) status resp.json()[data][run] if status DONE: return status if status in (FAIL, CANCEL): raise RuntimeError(fparse failed: {status}) time.sleep(5) raise TimeoutError(parse timeout) if __name__ __main__: dataset_id create_dataset(product-manual) doc_id upload_document(dataset_id, ./docs/产品手册.pdf) parse_document(dataset_id, doc_id) print(wait_until_ready(dataset_id, doc_id))6.2 Strands 工具封装与 Agent 运行# agent_app.py import os import requests from strands_sdk import Agent, tool RAGFLOW_URL os.getenv(RAGFLOW_URL, http://localhost:9380) RAGFLOW_API_KEY os.getenv(RAGFLOW_API_KEY, ragflow-xxxx) DATASET_ID os.getenv(DATASET_ID, your-dataset-id) HEADERS {Authorization: fBearer {RAGFLOW_API_KEY}} tool def query_knowledge_base(question: str, top_k: int 5) - str: 在企业知识库中检索与问题最相关的段落返回内容会标注来源文档和段落编号。 当用户问题涉及产品参数、操作流程、售后政策、内部规范时必须先调用本工具。 payload { question: question, dataset_ids: [DATASET_ID], top_k: top_k, similarity_threshold: 0.2, vector_similarity_weight: 0.3, } resp requests.post( f{RAGFLOW_URL}/api/v1/retrieval, headersHEADERS, jsonpayload, timeout30, ) resp.raise_for_status() chunks resp.json()[data][chunks] if not chunks: return 知识库中没有找到相关内容。 return \n.join( f[{c[doc_name]}#{c[chunk_id]}] {c[content]} for c in chunks ) agent Agent( modelos.getenv(AGENT_MODEL, your-llm-endpoint), api_keyos.getenv(AGENT_API_KEY), tools[query_knowledge_base], system_prompt( 你是企业内部知识助手。\n 当用户问题涉及产品参数、操作流程、售后政策、内部规范时\n 必须先调用 query_knowledge_base 获取信息再基于检索结果回答。\n 回答中必须使用 [来源:文档名#段落号] 标注依据\n 如果检索结果中没有答案直接说明未找到不要编造。\n 当用户只是闲聊、寒暄或者问题明显与知识库无关时不要调用检索工具。 ), ) def ask(user_msg: str, history: list | None None): response agent.chat(user_msg, historyhistory) print(response.content) return response if __name__ __main__: ask(R100 设备的整机保修期是多久)6.3 一次会话的运行输出实际跑出来的效果大概是这样的模型会先触发检索工具拿到带来源的片段后组织回答用户R100 设备的整机保修期是多久 助手[来源:产品手册#42] 根据产品手册第 42 页的参数表格R100 设备整机保修期为 12 个月 核心部件电池、电源模块保修期为 24 个月。 R100 设备的整机保修期为 12 个月核心部件保修期为 24 个月。注意回答里第一行就带了引用来源这是提示词约束的结果不是模型偶然行为。6.4 关于这个示例想提醒的三点第一DATASET_ID不要写死。如果知识库有多个建议在工具函数里改成接收knowledge_base参数让 Agent 根据问题内容决定查哪个库。第二生产环境一定要给 RAGFlow 和 Strands SDK 的调用加日志链路追踪。否则像检索超时和模型答非所问这类问题定位会非常痛苦。第三检索工具返回的内容不要原样全量塞给模型。如果 RAGFlow 返回的 chunk 过长建议在工具内部先做一次简单截断或段落压缩只保留与问题关键词最相关的部分否则上下文会被撑爆。我个人的体会是Agent 加 RAG 的项目里最容易翻车的地方往往不在模型和框架而在知识进库和工具调用边界这两个不起眼的环节。RAGFlow 把前者做得足够省心Strands SDK 把后者接得足够自然剩下真正需要你动脑子的就是一套干净的参数和一套清晰的提示词。等你把这套链路跑通之后再回头解决多知识库路由、权限隔离、增量更新这些进阶问题会顺手很多。