影刀RPA 向量数据库与RAG实战打造企业知识库自动问答作者林焱什么情况用公司有几百份产品手册、操作规范、制度文件客服和销售人员每天被问到相同的问题「这个型号的参数是多少」「退换货流程怎么走」「合同模板在哪里下载」传统的做法是建一个共享文件夹或飞书知识库让大家自己搜。但问题是——搜出来的结果经常不对口要么信息太旧要么关键词没对上。用影刀RPA RAG检索增强生成可以做一个自动问答机器人把企业文档向量化存入知识库用户提问时自动检索最相关的文档片段再交给大模型组织成自然语言回答。核心场景企业有大量文档需要智能检索和自动问答。怎么做第一步理解RAG的完整链路┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 文档预处理 │ → │ 向量化存入 │ → │ 用户提问时 │ → │ 大模型生成 │ │ (切分清洗) │ │ (embedding) │ │ 检索相关片段 │ │ 自然语言回答 │ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘核心思路不用每次都把全部文档塞给大模型太贵太慢而是先用向量检索找到最相关的几段文字只把这些喂给大模型。第二步文档预处理——把大文件切成小块RAG的质量很大程度上取决于「切得对不对」。切太大→检索不准切太小→上下文不够。importosimportreclassDocumentSplitter:文档分块器def__init__(self,chunk_size500,overlap100): chunk_size: 每块的字数上限 overlap: 相邻块的重叠字数防止关键信息被切断 self.chunk_sizechunk_size self.overlapoverlapdefsplit_text(self,text,title):将长文本切成若干块chunks[]start0text_lenlen(text)whilestarttext_len:endmin(startself.chunk_size,text_len)chunk_texttext[start:end]# 尽量在句号或换行处切断避免半个句子ifendtext_len:last_periodmax(chunk_text.rfind(。),chunk_text.rfind(\n))iflast_periodself.chunk_size*0.5:endstartlast_period1chunk_texttext[start:end]chunks.append({title:title,content:chunk_text.strip(),chunk_index:len(chunks)})startend-self.overlapifendself.overlapelseendreturnchunksdefprocess_folder(self,folder_path):处理文件夹中的所有txt和md文件all_chunks[]forfilenameinos.listdir(folder_path):ifnotfilename.endswith((.txt,.md)):continuefilepathos.path.join(folder_path,filename)try:withopen(filepath,r,encodingutf-8)asf:textf.read()# 去掉多余空行textre.sub(r\n{3,},\n\n,text)chunksself.split_text(text,titlefilename)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.modelSentenceTransformer(shibing624/text2vec-base-chinese)# 初始化ChromaDBself.clientchromadb.PersistentClient(pathdb_path)self.collection_namecompany_docs# 获取或创建集合try:self.collectionself.client.get_collection(self.collection_name)except:self.collectionself.client.create_collection(nameself.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[fdoc_{i}foriinrange(len(chunks))]# 批量生成向量embeddingsself.model.encode(texts).tolist()# 存入ChromaDB支持增量添加self.collection.add(embeddingsembeddings,documentstexts,metadatasmetas,idsids)print(f已添加{len(chunks)}条记录到知识库)defsearch(self,query,top_k5):检索最相关的文档片段query_embeddingself.model.encode([query]).tolist()resultsself.collection.query(query_embeddingsquery_embedding,n_resultstop_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.kbknowledge_base self.aiai_callerdefask(self,question):回答用户问题# 1. 检索相关文档matchesself.kb.search(question,top_k3)# 2. 拼接上下文context_parts[]fori,minenumerate(matches):context_parts.append(f[参考资料{i1}来源{m[title]}]\n{m[content]})context\n\n.join(context_parts)# 3. 构造promptsystem_prompt你是一个企业知识库助手。请根据提供的参考资料回答问题。 规则 1. 只用参考资料中的信息回答不要编造 2. 如果参考资料不足以回答问题明确说现有资料无法回答该问题 3. 回答要简洁准确引用时标注来源user_msgf参考资料{context}用户问题{question}# 4. 调用大模型生成回答answerself.ai.chat(system_prompt,user_msg,temperature0.3)return{answer:answer,references:[{title:m[title],content:m[content][:100]...}forminmatchesifm[score]0.5]}# 完整使用流程 # 1. 切分文档splitterDocumentSplitter(chunk_size500)chunkssplitter.process_folder(./company_docs)# 2. 建知识库kbKnowledgeBase(./knowledge_db)kb.add_documents(chunks)# 3. 问答# assistant RAGAssistant(kb, ai_caller)# result assistant.ask(退换货流程是怎样的)# print(result[answer])第五步在影刀中集成——打造问答机器人 影刀Python节点RAG问答接口 用法影刀流程中用户输入问题→Python节点调用RAG→输出回答 # 全局初始化只执行一次importjson# 影刀传入的变量user_questionquestionuser_question# 从影刀变量读入# 初始化首次运行较慢加载模型ifkbnotindir():kbKnowledgeBase(./knowledge_db)# ai_caller 用之前的AICaller类初始化aiAICaller(api_keyyour-key,base_urlhttps://...,modelqwen-turbo)assistantRAGAssistant(kb,ai)# 执行问答resultassistant.ask(question)# 输出给影刀outputjson.dumps(result,ensure_asciiFalse)print(output)有什么坑坑1embedding模型第一次下载巨慢SentenceTransformer(shibing624/text2vec-base-chinese)第一次运行要下载模型文件约400MB。在公司网络不好的环境可能要等半小时。解决方法提前在有网络的机器上下好模型复制到离线环境。或者用更小的模型如all-MiniLM-L6-v2只有80MB但中文效果差一些。坑2ChromaDB的数据文件不能跨机器复制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让企业文档从「堆在文件夹里」变成「随问随答」。关键链路是文档切分→向量化→检索→大模型生成。入门推荐用ChromaDBtext2vec轻量免费。企业级场景考虑加rerank提升准确率。