1.什么是RAGRAG其实是三个独立的单词也是执行的基本步骤R检索--retrieveA增强--AugmentG生成--Generate简单描述就是让AI先查询资料再去回答问题就是有依据的回答而不是随便回答就例如你问一个你们公司业务方面的问题那AI是不知道的这时候就需要去文档查询找到相关的内容然后检索回答。2.实现一个简单RAG# 手写一个最简单的 RAG 系统 - 不依赖任何框架 from openai import OpenAI import os from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlhttps://api.deepseek.com ) # 1. 文档库模拟公司知识库 documents [ {id: 1, content: 公司报销流程填写报销单 → 部门经理审批 → 财务审核 → 打款, keywords: [报销, 流程, 审批]}, {id: 2, content: 请假规则事假需提前3天申请病假可当天申请年假需提前1周, keywords: [请假, 事假, 病假, 年假]}, {id: 3, content: 加班政策工作日加班1.5倍工资周末加班2倍工资节假日3倍, keywords: [加班, 工资, 节假日]}, {id: 4, content: 电脑申请流程填写IT申请表 → IT部门审批 → 领取设备约3个工作日, keywords: [电脑, IT, 申请]}, ] # 2. 检索函数关键词匹配 def retrieve(query: str, top_k2): 根据用户问题检索相关文档 用关键词匹配模拟实际场景会用向量数据库 query_lower query.lower() scores [] for doc in documents: score 0 # 计算匹配度问题中的词在文档关键词中出现的次数 for word in query_lower.split(): for kw in doc[keywords]: if kw in word or word in kw: score 1 scores.append(score) # 按分数排序取 top_k 个 indexed list(enumerate(scores)) indexed.sort(keylambda x: x[1], reverseTrue) results [] for i in range(min(top_k, len(indexed))): idx indexed[i][0] if indexed[i][1] 0: results.append(documents[idx]) return results # 3. 生成回答基于检索结果 def generate_answer(query: str, contexts: list): 基于检索到的文档生成回答 if not contexts: return 抱歉没有找到相关信息。 # 构建上下文 context_text \n\n.join([f【参考文档{i1}】{doc[content]} for i, doc in enumerate(contexts)]) prompt f 你是一个企业助手。请根据以下参考文档回答用户问题。 【参考文档】 {context_text} 【用户问题】 {query} 【要求】 - 只根据参考文档回答 - 如果文档中没有相关信息请说文档中没有提到 - 回答要简洁准确 【回答】 response client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], temperature0 ) return response.choices[0].message.content # 4. RAG 主流程 def rag_ask(query: str): RAG 问答入口 print(f\n{*50}) print(f用户问题: {query}) print(f{*50}) # 步骤1检索相关文档 print( 正在检索相关文档...) contexts retrieve(query) print(f 找到 {len(contexts)} 个相关文档) for doc in contexts: print(f - {doc[content][:50]}...) # 步骤2基于检索结果生成回答 print( 正在生成回答...) answer generate_answer(query, contexts) return answer # 5. 测试 if __name__ __main__: print(\n *50) print(RAG 系统演示) print(*50) # 测试1报销问题 result rag_ask(怎么报销费用) print(f\n✅ 回答: {result}) # 测试2请假问题 result rag_ask(请事假要提前多久) print(f\n✅ 回答: {result}) # 测试3不在文档中的问题 result rag_ask(公司食堂几点开门) print(f\n✅ 回答: {result})3.解析代码我们直接从主程式看也就是rag_ask函数第一个问题是怎么报销费用回忆上面说的RAG步骤第一步需要去检索调用retrieve函数传入问题和拿到文档的个数限制程序遍历了documents文档然后用问题在keywords里面寻找相关的文档每次按照匹配上的次数多少给score 1然后进行编号元组、转成列表排序、取值indexed是[(2, 2), (0, 1), (3, 1), (1, 0)]min(top_k, len(indexed))取top_k和len(indexed)中较小的作为匹配的数量进行循环取到下标拿到documents中对应的文件作为结果第二步把拿到的文件给AI让AI在里面找答案做出回答answer generate_answer(query, contexts)这里面简单一些就是把doc的信息作为context_text上下文传入提示词中一定要注明仅在参考文档中找答案第三步返回answer打印结果4.向量化上面是没有用向量数据库的RAG就是在理解原理检索的时候后面会引入向量数据库就是把documents转成成向量并存入数据库然后在向量数据库里面检索。先记住一些词Embedding模型、向量相似度、chroma、sentence-transformers。下一章继续。