一、RAG检索增强生成概念RAG系统整体架构流程RAG的完整流程可以分为四个核心阶段1、数据准备阶段把你的文档、数据库等各种知识源准备好2、索引构建阶段把这些知识“消化”成机器能快速查找的格式3、检索阶段用户问问题时快速找到相关的知识4、生成阶段结合找到的知识和问题生成靠谱的答案RAG的全程是Retrieval-Augmented Generation中文叫“检索增强生成”。RAG技术通过将检索机制与生成模型结合使LLM能够从外部知识源获取相关文档并基于这些文档生成答案从而解决了传统LLM的知识过时缺乏领域知识和幻觉等问题。RAG的核心优势特性传统LLMRAG系统知识更新需要重新训练成本高只需要更新知识库成本低事实准确性容易产生幻觉基于检索到的真实文档领域适应需要微调模型添加领域文档即可可追溯性无法追溯来源可以标注来源和引用成本训练和部署成本高相对经济实惠举个实际例子假设你在做一个医疗问答系统用户问“阿司匹林的常见副作用是什么”传统LLM可能回答“阿司匹林可能导致胃部不适、恶心等。(但不确定是否完整或最新)”RAG系统的工作流程检索系统在医学数据库中搜索阿司匹林 副作用找到3篇相关医学文献提取关键信息胃肠道反应、出血风险、过敏反应等LLM基于这些文献生成答案根据医学文献阿司匹林的常见副作用包括胃肠道反应胃痛、恶心、消化不良出血风险增加过敏反应如荨麻疹参考来源[文献1]、[文献2]二、Indexing索引构建索引构建是RAG的“预处理”阶段就像你上学前需要先整理笔记、做好书签一样这个阶段要把原始文档转化成机器能快速检索的格式。第一步数据获取收集数据可能来自PDF、Word、Markdown文件网页Wili数据库邮件聊天记录Excel, CSV等结构化数据。实际案例某电商公司想做客服机器人他们需要收集产品说明文档PDF、常见问题FAQ网页、历史工单数据数据库、退换货政策Word文档第二步文档预处理Preprocessing拿到原始数据后需要“清洗”一下# 示例文档预处理代码 def preprocess_document(doc): # 1. 移除多余的空格和换行 doc re.sub(r\s, ,doc) # 2. 提取纯文本从pdf, HTML等 if doc_type pdf: text extract_text_from_pdf(doc) # 3. 规范化格式 text text.strip().lower() # 4. 去除无用信息页眉、页脚等 text remove_headers_footers(text) return text第三步 文档分块Chunking最关键的一步把大文档切成小段每段包含一个相对完整的语义单元常见的分块策略1、固定大小分块最简单但可能切断语义# 每300个字符一块 chunk_size 300 chunks [text[i:ichunk_size] for i in range(0, len(text),chunk_size)]2、句子分块按自然语言边界# 按句号、问号、感叹号分割 import nltk # 段落分块保留逻辑结构 sentences nltk.sent_tokenize(text) # 按换行符或者段落标记分割 chunks text.split(\n\n) # 滑动窗口分块带重叠、避免信息丢失 chunk_size 300 overlap 50 chunks [] for i in range(0, len(text), chunk_size - overlap): chunks.append(text[i:ichunk_size])分块的关键参数块大小Chunk Size太小 —— 语义不完整太大 —— 检索不精准推荐200 - 500 tokens (约150 - 400个汉字)重叠Overlap避免关键信息被切断推荐10-20%的块大小举个例子原文 阿司匹林是一种常用的解热镇痛药。它的主要作用包括 解热降低发烧体温 镇痛缓解轻到中度疼痛 抗血小板预防血栓形成 但是阿司匹林也有副作用。常见的副作用包括 胃肠道反应胃痛、恶心 出血风险特别是长期服用分块后Chunk 1: 阿司匹林是一种常用的解热镇痛药。它的主要作用包括 1. 解热降低发烧体温 2. 镇痛缓解轻到中度疼痛 3. 抗血小板预防血栓形成 Chunk 2: 阿司匹林的主要作用包括预防血栓形成。但是阿司匹林也有副作用。 常见的副作用包括 1. 胃肠道反应胃痛、恶心 2. 出血风险特别是长期服用注意chunk2有重叠这样即使用户搜“副作用”也能同时看到“作用”相关的上下文。from typing import List import re class TextSplitter: 文本分割器 def __init__( self, chunk_size: int 500, chunk_overlap: int 50, separators: List[str] None ): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.separtors separators or [\n\n,\n,。,! ,? , ] # 段落 - 换行 - 中文句子 - 单词 def split_text(self, text: str) - List[str]: 分割文本 chunks [] current_chunk # 按分隔符分割 sentences self._split_by_separators(text) for sentence in sentences: if len(current_chunk) len(sentence) self.chunk_size: current_chunk sentence else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk sentence if current_chunk: chunks.append(current_chunk.strip()) return chunks def _split_by_separators(self, text: str) - List[str]: 按分隔符分割 result [text] 第一轮按 \n\n 切 第二轮对结果再按 \n 切 第三轮再按 。 切 for separator in self.separators: # 逐级分割 new_result [] for text in result: new_reuslt.extend(text.split(separator)) resule new_result reurn [s for s in result if s.strip()] # 使用示例 def create_chunks(text: str, chunk_size: int 500) - List[str]: 创建文本块 splitter TextSplitter(chunk_size chunk_size) return splitter.split_text(text)第四步 向量化Embedding分块后的文本还是人类语言机器不懂。我们需要把它转换成向量一串数字。什么是Embedding?把文字转换成数字向量相似的文字会得到相似的向量。#使用OpenAI的Embedding模型 from openai import OpenAI client OpenAI() text 阿司匹林是一种解热镇痛药 response client.embeddings.create( modeltext-embedding-3-small, inputtext ) vector response.data[0].embedding print(f向量维度: {len(vector)}) # 输出: 1536 print(f前5个值: {vector[:5]}) # 输出: [0.023, -0.014, 0.089, ...]向量化之后向量之间可以计算相似度余弦相似度数值越接近1越相似感冒药 → [0.1, 0.3, 0.5, ...] 抗感冒药物 → [0.12, 0.29, 0.51, ...] # 向量很接近 汽车 → [0.8, 0.1, 0.2, ...] # 向量差很远 from sklearn.metrics.pairwise import cosine_similarity import numpy as np vec1 np.array([0.1, 0.3, 0.5]) vec2 np.array([0.12, 0.29, 0.51]) similarity cosine_similarity([vec1], [vec2])[0][0] print(f相似度: {similarity:.3f}) # 输出: 0.999非常相似常用的Embedding模型模型维度优势适用场景OpenAI text-embedding-3-small1536效果好速度快通用场景OpenAI text-embedding-3-large3072效果最好高精度需求BGE-large-zh1024中文效果好中文为主Sentence-BERT768开源免费资源有限向量嵌入的完整实现import openai import AsyncOpenAI from typing import List import numpy as np async def get_embeddings( texts: List[str], model: str text-embedding-3-small ) - List[List[float]]: client AsyncOpenAI() # 读取API Key, 通常来自.env;建立连接 response await client.embeddings.create( model model, input texts ) # 将文本送入模型返回向量表示 return [item.embedding for item in response.data] # 相似度计算 def cosine_similarity(a: List[float], b:List[float]) - float: 计算余弦相似度 a np.array(a) b np.array(b) return np.dot(a, b)/(np.linalg.norm(a)*np.linalg.norm(b)) # 使用示例 async def find_similar(query: str, documents: List[str], top_k: int 3): 找到最相似的文档 # 获取查询向量 query_embedding (await get_embedding([query]))[0] # 获取文档向量 doc_embedding await get_embeddings(documents) # 计算相似度 similarities [ (i, cosine_similarity(query_embedding, doc_emb)) for i, doc_emb in enmuerate(doc_embeddings) ] # 排序并返回top_k similarities.sort(keylambda x:x[1], reverse True) return [(documents[i],sim) for i,sim in similarities[:top_k]]第五步 存储到向量数据库最后把这些向量存起来用专门的向量数据库Vector Database。为什么不用普通数据库普通数据库MySQL、MongoDB擅长精确查询“找ID 123的记录”但是向量搜索是相似性查询“找和[0.1, 0.3, 0.5]最相似的10个向量“。向量数据库用了特殊的索引算法如HNSW、IVF,能在百万、千万级向量中毫秒级找到最相似的。常用向量数据库1、Pinecone (云服务简单好用)import pinecone pinecone.init(api_key your-api-key) index pinecone.Index(my-rag-index) # 插入向量 index.upsert( [ (doc1_chunk1, vector1, {text:阿司匹林是...}), (doc1_chunk2, vector2, {text:副作用包括...}) ] ) # 查询 results index.query(query_vector, top_k 3)2、Milvus (开源功能强大)3、FAISSFacebook开源本地使用4、Weaviate支持混合搜索完整的索引构建代码示例# # 1. 导入 # from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # # 2. 加载文档 # loader TextLoader(data.txt, encodingutf-8) documents loader.load() print(f加载文档数: {len(documents)}) # # 3. 文档分块核心 # text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , ] ) # 核心算法优先大语义 → 不行再细分 → 最后强制切 chunks text_splitter.split_documents(documents) print(f分块数量: {len(chunks)}) # # 4. 向量化模型 # embeddings OpenAIEmbeddings( modeltext-embedding-3-small ) # # 5. 构建向量数据库 # # 内部结构id → embedding → text → metadata vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db ) # 通过pinecone构建 pinecone.init(api_keyyour-key) index_name medical-rag #5. 创建索引并存储 vectorstore Pinecone.from_texts( textschunks, embeddingembeddings, index_nameindex_name ) # # 6. 持久化 # vectorstore.persist()三、Retrieval 检索快速找到相关知识检索阶段的任务是从海量知识中快速找出最相关的那几条查询处理用户输入的问题往往不是完美的搜索词。比如用户原始问题我吃了阿司匹林后胃疼怎么办 需要优化为 阿司匹林 胃痛 副作用关键词提取 阿司匹林导致的胃部不适如何处理问题改写向量检索1、Vector Retriever: 基于Embedding。能理解语义但是对专有名词数字等不敏感。# 加载数据库 from langchain_core.documents import Document docs [ Document(page_contentE.coli is a common pathogen in lung infection, metadata{species: E.coli}), Document(page_contentStreptococcus pneumoniae causes pneumonia, metadata{species: S.pneumoniae}), Document(page_contentNormal lung microbiome contains commensal bacteria, metadata{type: commensal}), ] from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 向量化 embedding OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma.from_documents( docs, embedding, persist_directory./db ) ########### 构建retriever ########### vector_retriever vectorstore.as_retriever( search_kwargs{k: 3} ) results vector_retriever.invoke(lung infection bacteria)2、Keyword Retriever (关键词/BM25): 传统搜索基于关键词匹配########### Keyword Retriever (BM25) ########### from langchain_community.retrievers import BM25Retriever bm25_retriever BM25Retriever.from_documents(docs) bm25_retriever.k 3 results bm25_retriever.invoke(E.coli infection)3、Hybrid Retriever检索系统通常采用混合搜索方法结合向量搜索找语义相似的文档和关键词搜索精确匹配然后对结果进行排序和过滤。初步检索后可以用更精细的模型重新排序提高精度。########### Hybrid Retriever ########### import numpy as np class HybridRetriever: def __init__(self, retrievers, weightsNone): self.retrievers retrievers self.weights weights or [1.0] * len(retrievers) def invoke(self, query): all_docs [] scores [] for retriever, weight in zip(self.retrievers, self.weights): docs retriever.invoke(query) for i, doc in enumerate(docs): # 简单排名分数可换更复杂 score weight * (1.0 / (i 1)) all_docs.append(doc) scores.append(score) # 排序 ranked sorted(zip(all_docs, scores), keylambda x: x[1], reverseTrue) # 去重 seen set() result [] for doc, _ in ranked: if doc.page_content not in seen: seen.add(doc.page_content) result.append(doc) return result hybrid HybridRetriever( retrievers[vector_retriever, bm25_retriever], weights[0.7, 0.3] ) results hybrid.invoke(E.coli infection)四、Generation (生成基于知识产生答案)检索到相关文档后最后一步就是生成答案提示词的构建Prompt Engineering把检索到的文档和用户问题组合成一个完整的提示词Prompt,发给LLM基础模版prompt_template 你是一个专业的医疗助手。请基于以下参考资料回答用户的问题。 如果参考资料中没有相关信息请明确告知用户。 参考资料 {context} 用户问题 {question} 请提供详细、准确的回答并标注信息来源。 #填充内容 context \n\n.join([doc.page_content for doc in retrieved_docs]) question 阿司匹林有哪些副作用 prompt prompt_template.format(contextcontext, questionquestion)进阶模版结构化输出advanced_prompt 参考资料 {context} 用户问题{question} 请按以下格式回答 直接回答用1-2句话概括答案 详细说明展开解释分点列出 注意事项相关的警告或建议 信息来源标注参考了哪些资料 回答 调用LLM生成from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4, messages[ {role: system, content: 你是一个专业的医疗助手}, {role: user, content: prompt} ], temperature0.3, # 降低随机性提高准确性 max_tokens500 ) answer response.choices[0].message.content print(answer)答案后处理生成答案后可能还需要1、引用标注给答案加上来源链接def add_citations(answer, sources): 给答案添加引用 for i, source in enumerate(sources): answer f\n[{i1}] {source[title]} - {source[url]} return answer2、安全过滤检查答案是否包含有害内容def safety_check(answer): harmful_keywords [自杀,违法,...] for keyword in harmful_keywords: if keyword in answer: return 抱歉这个问题涉及敏感内容建议咨询专业人士 return answer完整的RAG代码def rag_query(question): 完整的RAG查询流程 # 1. 检索相关文档 retrieved_docs vectorstore.similarity_search(question, k3) # 2. 构建prompt context \n\n.join([ f【文档{i1}】{doc.page_content} for i, doc in enumerate(retrieved_docs) ]) prompt f 参考以下资料回答问题 {context} 问题{question} 要求 1. 回答要准确、专业 2. 必须基于参考资料 3. 标注信息来源 # 3. 调用LLM生成 response client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}], temperature0.3 ) answer response.choices[0].message.content # 4. 添加引用 sources [ {title: f文档{i1}, score: doc.metadata.get(score, 0)} for i, doc in enumerate(retrieved_docs) ] return { answer: answer, sources: sources, retrieved_docs: [doc.page_content for doc in retrieved_docs] } #使用示例 result rag_query(阿司匹林有哪些副作用) print(答案:, result[answer]) print(\n参考来源:, result[sources])01什么是AI大模型应用开发工程师如果说AI大模型是蕴藏着巨大能量的“后台超级能力”那么AI大模型应用开发工程师就是将这种能量转化为实用工具的执行者。AI大模型应用开发工程师是基于AI大模型设计开发落地业务的应用工程师。这个职业的核心价值在于打破技术与用户之间的壁垒把普通人难以理解的算法逻辑、模型参数转化为人人都能轻松操作的产品形态。无论是日常写作时用到的AI文案生成器、修图软件里的智能美化功能还是办公场景中的自动记账工具、会议记录用的语音转文字APP这些看似简单的应用背后都是应用开发工程师在默默搭建技术与需求之间的桥梁。他们不追求创造全新的大模型而是专注于让已有的大模型“听懂”业务需求“学会”解决具体问题最终形成可落地、可使用的产品。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】02AI大模型应用开发工程师的核心职责需求分析与拆解是工作的起点也是确保开发不偏离方向的关键。应用开发工程师需要直接对接业务方深入理解其核心诉求——不仅要明确“要做什么”更要厘清“为什么要做”以及“做到什么程度算合格”。在此基础上他们会将模糊的业务需求拆解为具体的技术任务明确每个环节的执行标准并评估技术实现的可行性同时定义清晰的核心指标为后续开发、测试提供依据。这一步就像建筑前的图纸设计若出现偏差后续所有工作都可能白费。技术选型与适配是衔接需求与开发的核心环节。工程师需要根据业务场景的特点选择合适的基础大模型、开发框架和工具——不同的业务对模型的响应速度、精度、成本要求不同选型的合理性直接影响最终产品的表现。同时他们还要对行业相关数据进行预处理通过提示词工程优化模型输出或在必要时进行轻量化微调让基础模型更好地适配具体业务。此外设计合理的上下文管理规则确保模型理解连贯需求建立敏感信息过滤机制保障数据安全也是这一环节的重要内容。应用开发与对接则是将方案转化为产品的实操阶段。工程师会利用选定的开发框架构建应用的核心功能同时联动各类外部系统——比如将AI模型与企业现有的客户管理系统、数据存储系统打通确保数据流转顺畅。在这一过程中他们还需要配合设计团队打磨前端交互界面让技术功能以简洁易懂的方式呈现给用户实现从技术方案到产品形态的转化。测试与优化是保障产品质量的关键步骤。工程师会开展全面的功能测试找出并修复开发过程中出现的漏洞同时针对模型的响应速度、稳定性等性能指标进行优化。安全合规性也是测试的重点需要确保应用符合数据保护、隐私安全等相关规定。此外他们还会收集用户反馈通过调整模型参数、优化提示词等方式持续提升产品体验让应用更贴合用户实际使用需求。部署运维与迭代则贯穿产品的整个生命周期。工程师会通过云服务器或私有服务器将应用部署上线并实时监控运行状态及时处理突发故障确保应用稳定运行。随着业务需求的变化他们还需要对应用功能进行迭代更新同时编写完善的开发文档和使用手册为后续的维护和交接提供支持。03薪资情况与职业价值市场对这一职业的高度认可直接体现在薪资待遇上。据猎聘最新在招岗位数据显示AI大模型应用开发工程师的月薪最高可达60k。在AI技术加速落地的当下这种“技术业务”的复合型能力尤为稀缺让该职业成为当下极具吸引力的就业选择。AI大模型应用开发工程师是AI技术落地的关键桥梁。他们用专业能力将抽象的技术转化为具体的产品让大模型的价值真正渗透到各行各业。随着AI场景化应用的不断深化这一职业的重要性将更加凸显也必将吸引更多人才投身其中推动AI技术更好地服务于社会发展。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】