影刀RPA 向量数据库与RAG实战：打造企业知识库自动问答

作者：林焱

什么情况用

公司有几百份产品手册、操作规范、制度文件，客服和销售人员每天被问到相同的问题：「这个型号的参数是多少？」「退换货流程怎么走？」「合同模板在哪里下载？」

传统的做法是建一个共享文件夹或飞书知识库，让大家自己搜。但问题是——搜出来的结果经常不对口，要么信息太旧，要么关键词没对上。

用影刀RPA + RAG（检索增强生成）可以做一个自动问答机器人：把企业文档向量化存入知识库，用户提问时自动检索最相关的文档片段，再交给大模型组织成自然语言回答。

核心场景：企业有大量文档需要智能检索和自动问答。

怎么做

第一步：理解RAG的完整链路

┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 文档预处理 │ → │ 向量化存入 │ → │ 用户提问时 │ → │ 大模型生成 │ │ (切分清洗) │ │ (embedding) │ │ 检索相关片段 │ │ 自然语言回答 │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘

核心思路：不用每次都把全部文档塞给大模型（太贵太慢），而是先用向量检索找到最相关的几段文字，只把这些喂给大模型。

第二步：文档预处理——把大文件切成小块

RAG的质量很大程度上取决于「切得对不对」。切太大→检索不准；切太小→上下文不够。

importosimportreclassDocumentSplitter:"""文档分块器"""def__init__(self,chunk_size=500,overlap=100):""" chunk_size: 每块的字数上限 overlap: 相邻块的重叠字数（防止关键信息被切断） """self.chunk_size=chunk_size self.overlap=overlapdefsplit_text(self,text,title=""):"""将长文本切成若干块"""chunks=[]start=0text_len=len(text)whilestart<text_len:end=min(start+self.chunk_size,text_len)chunk_text=text[start:end]# 尽量在句号或换行处切断，避免半个句子ifend<text_len:last_period=max(chunk_text.rfind("。"),chunk_text.rfind("\n"))iflast_period>self.chunk_size*0.5:end=start+last_period+1chunk_text=text[start:end]chunks.append({"title":title,"content":chunk_text.strip(),"chunk_index":len(chunks)})start=end-self.overlapifend>self.overlapelseendreturnchunksdefprocess_folder(self,folder_path):"""处理文件夹中的所有txt和md文件"""all_chunks=[]forfilenameinos.listdir(folder_path):ifnotfilename.endswith(('.txt','.md')):continuefilepath=os.path.join(folder_path,filename)try:withopen(filepath,"r",encoding="utf-8")asf:text=f.read()# 去掉多余空行text=re.sub(r'\n{3,}','\n\n',text)chunks=self.split_text(text,title=filename)all_chunks.extend(chunks)exceptExceptionase:print(f"处理{filename}失败：{e}")print(f"共处理{len(all_chunks)}个文本块")returnall_chunks

第三步：向量化——把文字变成数字

文字没法直接做相似度计算，需要先转换成向量。这里用text2vec-base-chinese模型，免费开源。

""" 先安装依赖： pip install sentence-transformers chromadb """fromsentence_transformersimportSentenceTransformerimportchromadbimportnumpyasnpclassKnowledgeBase:"""基于ChromaDB的向量知识库"""def__init__(self,db_path="./knowledge_db"):# 加载中文embedding模型self.model=SentenceTransformer("shibing624/text2vec-base-chinese")# 初始化ChromaDBself.client=chromadb.PersistentClient(path=db_path)self.collection_name="company_docs"# 获取或创建集合try:self.collection=self.client.get_collection(self.collection_name)except:self.collection=self.client.create_collection(name=self.collection_name,metadata={"hnsw:space":"cosine"}# 用余弦相似度)defadd_documents(self,chunks):"""将文档块加入知识库"""texts=[c["content"]forcinchunks]metas=[{"title":c["title"],"chunk_index":c["chunk_index"]}forcinchunks]ids=[f"doc_{i}"foriinrange(len(chunks))]# 批量生成向量embeddings=self.model.encode(texts).tolist()# 存入ChromaDB（支持增量添加）self.collection.add(embeddings=embeddings,documents=texts,metadatas=metas,ids=ids)print(f"已添加{len(chunks)}条记录到知识库")defsearch(self,query,top_k=5):"""检索最相关的文档片段"""query_embedding=self.model.encode([query]).tolist()results=self.collection.query(query_embeddings=query_embedding,n_results=top_k)# 格式化返回结果matches=[]foriinrange(len(results["documents"][0])):matches.append({"content":results["documents"][0][i],"title":results["metadatas"][0][i].get("title",""),"score":1-results["distances"][0][i]# 转成相似度分数})returnmatches

第四步：组装RAG问答

检索到相关文档片段后，拼成prompt发给大模型。

classRAGAssistant:"""RAG问答助手"""def__init__(self,knowledge_base,ai_caller):self.kb=knowledge_base self.ai=ai_callerdefask(self,question):"""回答用户问题"""# 1. 检索相关文档matches=self.kb.search(question,top_k=3)# 2. 拼接上下文context_parts=[]fori,minenumerate(matches):context_parts.append(f"[参考资料{i+1}来源：{m['title']}]\n{m['content']}")context="\n\n".join(context_parts)# 3. 构造promptsystem_prompt="""你是一个企业知识库助手。请根据提供的参考资料回答问题。 规则： 1. 只用参考资料中的信息回答，不要编造 2. 如果参考资料不足以回答问题，明确说"现有资料无法回答该问题" 3. 回答要简洁准确，引用时标注来源"""user_msg=f"""参考资料：{context}用户问题：{question}"""# 4. 调用大模型生成回答answer=self.ai.chat(system_prompt,user_msg,temperature=0.3)return{"answer":answer,"references":[{"title":m["title"],"content":m["content"][:100]+"..."}forminmatchesifm["score"]>0.5]}# ===== 完整使用流程 =====# 1. 切分文档splitter=DocumentSplitter(chunk_size=500)chunks=splitter.process_folder("./company_docs")# 2. 建知识库kb=KnowledgeBase("./knowledge_db")kb.add_documents(chunks)# 3. 问答# assistant = RAGAssistant(kb, ai_caller)# result = assistant.ask("退换货流程是怎样的？")# print(result["answer"])

第五步：在影刀中集成——打造问答机器人

""" 影刀Python节点：RAG问答接口 用法：影刀流程中，用户输入问题→Python节点调用RAG→输出回答 """# 全局初始化（只执行一次）importjson# 影刀传入的变量：user_questionquestion=user_question# 从影刀变量读入# 初始化（首次运行较慢，加载模型）if"kb"notindir():kb=KnowledgeBase("./knowledge_db")# ai_caller 用之前的AICaller类初始化ai=AICaller(api_key="your-key",base_url="https://...",model="qwen-turbo")assistant=RAGAssistant(kb,ai)# 执行问答result=assistant.ask(question)# 输出给影刀output=json.dumps(result,ensure_ascii=False)print(output)

有什么坑

坑1：embedding模型第一次下载巨慢

SentenceTransformer("shibing624/text2vec-base-chinese")第一次运行要下载模型文件，约400MB。在公司网络不好的环境，可能要等半小时。

解决方法：提前在有网络的机器上下好模型，复制到离线环境。或者用更小的模型（如all-MiniLM-L6-v2，只有80MB，但中文效果差一些）。

坑2：ChromaDB的数据文件不能跨机器复制

ChromaDB的持久化数据依赖SQLite，如果直接复制文件夹到另一台机器，可能因为路径或权限问题读不出来。

踩坑实录：在开发机上建好知识库，直接打包复制到生产服务器，启动时报sqlite3.OperationalError: database is locked。后来发现是ChromaDB版本不一致导致的。

解决方法：用版本锁定的requirements.txt，或在目标机器上重新执行add_documents建库。

坑3：文档切分不合理导致回答质量差

切得太碎（比如每100字一块），检索精度高但上下文不足，大模型看不懂。切得太粗（每2000字一块），检索相关性差，喂给大模型的都是无关内容。

实际经验：500字一块 + 100字重叠，对于中文产品手册效果最好。如果是合同这类结构化文档，建议按条款切分，不要机械按字数。

坑4：向量检索的"语义漂移"

用户问「这款产品能退货吗？」向量检索可能返回关于「产品质量标准」的内容，因为语义上有重叠，但实际上不是用户要的。

解决方法：提高top_k到10，然后加一道重排序（rerank）——用大模型对检索结果做二次筛选。代价是成本翻倍，准确率也翻倍。

坑5：企业文档更新频率

产品手册改了、价格表更新了，如果知识库不更新，问答机器人就会用旧数据回答。

解决方法：在影刀里建一个定时流程——每天凌晨扫描文档文件夹，发现有变更的文件就重新向量化，增量更新到ChromaDB。

总结：RAG让企业文档从「堆在文件夹里」变成「随问随答」。关键链路是：文档切分→向量化→检索→大模型生成。入门推荐用ChromaDB+text2vec，轻量免费。企业级场景考虑加rerank提升准确率。