欢迎来到人工智能的世界博客主页卿云阁欢迎关注点赞收藏⭐️留言首发时间2026年6月1日✉️希望可以和大家一起完成进阶之路作者水平很有限如果发现错误请留言轰炸哦万分感谢目录Embedding、向量库和 RAG文本嵌入模型 Embedding向量库 FAISSRetriever 检索器RAG检索增强生成RAG 的标准 LangChain完整 RAG 标准流程SQL 链和综合小项目SQLDatabase 连接数据库 手动执行 SQL大语言模型生成和执行SQL使用指定的数据表Embedding、向量库和 RAG文本嵌入模型 EmbeddingEmbedding 就是把文字变成一串数字向量让程序能计算“文字之间像不像”。“苹果手机”和“iPhone”很像“苹果手机”和“红烧牛肉面”不像from dotenv import load_dotenv import os from langchain_openai import OpenAIEmbeddings load_dotenv() embeddings OpenAIEmbeddings( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeltext-embedding-v4, dimensions1024, # 关键关闭 OpenAIEmbeddings 内部的长度检查/token 处理 check_embedding_ctx_lengthFalse ) text LangChain 是一个开发大模型应用的框架。 vector embeddings.embed_query(text) print(向量长度) print(len(vector)) print(向量前 10 个数字) print(vector[:10])向量库 FAISSEmbedding负责将自然语言文本转化为计算机可识别的数字向量实现文本向量化FAISS是专门存储这些向量并支持相似度检索的向量数据库。在 RAG 流程中先将多篇文档通过 Embedding转为向量并存入FAISS用户提问时先对问题向量化再通过FAISS的similarity_search检索出与问题最相似、最相关的文本片段同时借助LangChain快速搭建大模型应用搭配PromptTemplate制作可复用的提示词最终实现基于私有知识库的智能问答效果。from dotenv import load_dotenv import os from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS load_dotenv() # 1. 创建 Embedding 模型 embeddings OpenAIEmbeddings( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeltext-embedding-v4, dimensions1024, check_embedding_ctx_lengthFalse ) # 2. 准备几段文本 texts [ LangChain 可以用来开发大模型应用。, FAISS 是一个向量数据库可以做相似度搜索。, PromptTemplate 用来创建可复用的提示词。 ] # 3. 把文本变成向量并存入 FAISS vectorstore FAISS.from_texts( textstexts, embeddingembeddings ) # 4. 提出一个问题 query 什么工具可以做相似度搜索 # 5. 搜索最相似的文本 docs vectorstore.similarity_search(query, k1) # 6. 打印结果 print(问题) print(query) print(最相关的内容) print(docs[0].page_content)FAISS 向量库默认仅存在于内存中程序结束后会消失使用vectorstore.save_local(faiss_index) 可将其保存到本地磁盘生成 index.faiss 和 index.pkl文件后续通过 FAISS.load_local 即可重新加载使用。Retriever 检索器通过vectorstore.as_retriever()可将 FAISS 向量库转为 LangChain 标准检索器。它本质是对向量库检索能力的统一封装和 FAISS 原生的similarity_search功能效果相近但检索器是LangChain 生态的通用接口搭建 RAG 应用时更推荐使用流程为向量库转为检索器再为RAG 流程提供相关文档检索能力。from dotenv import load_dotenv import os from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS load_dotenv() # 1. 创建 Embedding 模型 embeddings OpenAIEmbeddings( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeltext-embedding-v4, dimensions1024, check_embedding_ctx_lengthFalse ) # 2. 准备文本 texts [ LangChain 是一个用于开发大模型应用的框架。, FAISS 是一个向量数据库可以用来做相似度搜索。, PromptTemplate 用来创建可复用的提示词模板。, Retriever 是检索器用来根据问题找到相关文档。 ] # 3. 创建 FAISS 向量库 vectorstore FAISS.from_texts( textstexts, embeddingembeddings ) # 4. 把向量库变成 Retriever 检索器 retriever vectorstore.as_retriever( search_kwargs{k: 2} ) # 5. 使用 retriever 检索 query 什么工具可以根据问题找到相关文档 docs retriever.invoke(query) # 6. 打印结果 print(问题) print(query) print(\n检索到的文档) for index, doc in enumerate(docs, start1): print(f\n第 {index} 条) print(doc.page_content)RAG检索增强生成RAG 即检索增强生成完整流程依托三大核心组件实现先用 Embedding 将文本转为向量存入FAISS 向量库再通过 Retriever 检索出与用户问题相关的文档。借助create_stuff_documents_chain可把检索到的文档内容填充至提示词的{context}占位符中最终交由大模型结合参考资料作答。最简实践流程为本地文本→向量化存入 FAISS→转为检索器→检索相关内容→大模型生成回答。本地文本 - FAISS - Retriever - Prompt - Model - 回答from dotenv import load_dotenv import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser load_dotenv() # 1. 创建大模型 model ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen3-vl-flash ) # 2. 创建 Embedding 模型 embeddings OpenAIEmbeddings( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeltext-embedding-v4, dimensions1024, check_embedding_ctx_lengthFalse ) # 3. 准备本地知识库文本 texts [ LangChain 是一个用于开发大模型应用的框架。, FAISS 是一个向量数据库可以用来做相似度搜索。, Retriever 是检索器用来根据问题找到相关文档。, RAG 的意思是检索增强生成它会先检索资料再让大模型根据资料回答。 ] # 4. 创建 FAISS 向量库 vectorstore FAISS.from_texts( textstexts, embeddingembeddings ) # 5. 创建 Retriever 检索器 retriever vectorstore.as_retriever( search_kwargs{k: 2} ) # 6. 创建 RAG 提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的问答助手。请只根据给定资料回答问题不要编造。), (user, 已知资料 {context} 用户问题 {question} ) ]) # 7. 提问 question RAG 是什么意思 # 8. 先用 retriever 检索相关文档 docs retriever.invoke(question) # 9. 把检索到的文档合并成 context context \n.join([doc.page_content for doc in docs]) # 10. 创建链 chain prompt | model | StrOutputParser() # 11. 调用链让模型根据资料回答 result chain.invoke({ context: context, question: question }) # 12. 打印结果 print(检索到的资料) print(context) print(\n模型回答) print(result)RAG 的标准 LangChain在标准 RAG 流程中create_stuff_documents_chain() 是 LangChain 提供的封装函数用于自动将检索到的多篇文档填充到提示词的 {context} 位置替代手动拼接文档的操作。我们只需要传入大模型和提示词即可创建文档链调用时直接传入文档列表和用户问题就能让模型基于上下文生成回答。from dotenv import load_dotenv import os from langchain_openai import ChatOpenAI from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_classic.chains.combine_documents import create_stuff_documents_chain load_dotenv() model ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen3-vl-flash ) docs [ Document(page_contentLangChain 是一个用于开发大模型应用的框架。), Document(page_contentRAG 的意思是检索增强生成。), Document(page_contentRAG 会先检索相关资料再让大模型根据资料回答问题。) ] prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的问答助手。请只根据给定资料回答问题。), (user, 已知资料 {context} 问题 {input} ) ]) document_chain create_stuff_documents_chain( llmmodel, promptprompt ) result document_chain.invoke({ context: docs, input: RAG 是什么意思 }) print(result)document_chain create_stuff_documents_chain( llmmodel, promptprompt )把 docs 这些 Document 自动填进 prompt 里的 {context}然后交给 model 回答result document_chain.invoke({ context: docs, input: RAG 是什么意思 })普通 RAG 手动版Document - 自己 join 成字符串 - 塞进 promptcreate_stuff_documents_chainDocument - 自动塞进 prompt完整 RAG 标准流程将检索器与文档链结合即可实现LangChain标准完整RAG流程用户提出问题后先由Retriever从向量库中检索相关文档再通过create_stuff_documents_chain创建的文档链自动将检索结果填入提示词上下文最终交由大模型结合资料生成答案。其中Retriever承担资料检索工作文档链负责整合内容并完成问答。文本资料 - Embedding - FAISS - Retriever - Document Chain - 模型回答from dotenv import load_dotenv import os from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_classic.chains.combine_documents import create_stuff_documents_chain load_dotenv() # 1. 创建大模型 model ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen3-vl-flash ) # 2. 创建 Embedding 模型 embeddings OpenAIEmbeddings( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modeltext-embedding-v4, dimensions1024, check_embedding_ctx_lengthFalse ) # 3. 准备知识库文本 texts [ LangChain 是一个用于开发大模型应用的框架。, FAISS 是一个向量数据库可以用来做相似度搜索。, Retriever 是检索器用来根据问题找到相关文档。, RAG 的意思是检索增强生成它会先检索资料再让大模型根据资料回答。 ] # 4. 创建 FAISS 向量库 vectorstore FAISS.from_texts( textstexts, embeddingembeddings ) # 5. 创建 Retriever retriever vectorstore.as_retriever( search_kwargs{k: 2} ) # 6. 创建 Prompt prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的问答助手。请只根据给定资料回答问题不要编造。), (user, 已知资料 {context} 问题 {input} ) ]) # 7. 创建文档链 document_chain create_stuff_documents_chain( llmmodel, promptprompt ) # 8. 用户问题 question RAG 是什么意思 # 9. Retriever 先检索相关文档 docs retriever.invoke(question) # 10. 文档链根据检索到的文档回答 result document_chain.invoke({ context: docs, input: question }) # 11. 打印检索到的资料 print(检索到的资料) for index, doc in enumerate(docs, start1): print(f\n第 {index} 条) print(doc.page_content) # 12. 打印模型回答 print(\n模型回答) print(result)SQL 链和综合小项目SQLDatabase 连接数据库 手动执行 SQL此前学习的 Prompt、OutputParser、Chain 和 RAG 主要用于处理**非结构化的文本资料而 SQL链专门处理数据库中的结构化数据如学生表、订单表等。通过学习 SQLDatabase 连接数据库与 create_sql_query_chain() 生成 SQL 语句可实现自然语言转数据库查询用户用中文提问模型自动生成 SQL程序执行后返回真实数据结果。与 RAG 检索文档不同SQL 链是直接查询数据库必须先用 SQLDatabase.from_uri() 连接数据库让模型获取表结构、字段等信息才能正确生成 SQL。这是让大模型从聊天工具变成能对接真实业务数据、完成实际工作的核心能力。第一步创建一个 SQLite 数据库文件。import sqlite3 # 1. 连接 SQLite 数据库 # 如果 demo.db 不存在会自动创建 conn sqlite3.connect(demo.db) # 2. 创建游标 cursor conn.cursor() # 3. 创建学生表 cursor.execute( CREATE TABLE IF NOT EXISTS students ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER, class_name TEXT, math_score INTEGER ) ) # 4. 清空旧数据避免重复插入 cursor.execute(DELETE FROM students) # 5. 插入几条学生数据 students [ (小明, 18, 一班, 90), (小红, 17, 一班, 95), (小刚, 18, 二班, 82), (小丽, 17, 二班, 88) ] cursor.executemany( INSERT INTO students (name, age, class_name, math_score) VALUES (?, ?, ?, ?) , students) # 6. 提交保存 conn.commit() # 7. 关闭数据库连接 conn.close() print(SQLite 数据库 demo.db 创建成功)第二步用 LangChain 连接这个数据库。from langchain_community.utilities import SQLDatabase # 1. 连接 SQLite 数据库 db SQLDatabase.from_uri(sqlite:///demo.db) # 2. 查看数据库类型 print(数据库类型) print(db.dialect) # 3. 查看有哪些表 print(数据表) print(db.get_usable_table_names()) # 4. 手动执行 SQL result db.run(SELECT count(*) FROM students;) print(学生总数) print(result)大语言模型生成和执行SQLfrom dotenv import load_dotenv import os from langchain_openai import ChatOpenAI from langchain_community.utilities import SQLDatabase from langchain_classic.chains import create_sql_query_chain load_dotenv() db SQLDatabase.from_uri(sqlite:///demo.db) model ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen3-vl-flash ) chain create_sql_query_chain( llmmodel, dbdb ) question 一共有多少个学生只返回 SQL 语句不要返回 SQLQuery: sql chain.invoke({ question: question }) print(模型原始输出) print(sql) # 关键清理掉 SQLQuery: sql sql.replace(SQLQuery:, ).strip() print(\n清理后的 SQL) print(sql) result db.run(sql) print(\n数据库查询结果) print(result)大模型本身并不知道数据库的表结构无法直接生成正确的SQL语句。通过SQLDatabase.from_uri()连接数据库后LangChain会自动获取数据库类型、表名、字段等schema信息并由create_sql_query_chain()将这些结构拼接进提示词传递给模型模型依据这些真实结构生成合法、正确的SQL查询命令而非凭空猜测。使用指定的数据表在使用 SQL 链时可以通过 table_names_to_use 参数指定模型只能使用哪些数据表例如限制仅查询 students 表而忽略数据库中的其他表。这样做能让模型只参考指定表的结构生成SQL有效减少误用表的情况、缩短提示词长度、提升查询准确性与安全性。from dotenv import load_dotenv import os from langchain_openai import ChatOpenAI from langchain_community.utilities import SQLDatabase from langchain_classic.chains import create_sql_query_chain load_dotenv() # 1. 连接 SQLite 数据库 db SQLDatabase.from_uri(sqlite:///demo.db) # 2. 创建大模型 model ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen3-vl-flash ) # 3. 创建 SQL 查询链 chain create_sql_query_chain( llmmodel, dbdb ) # 4. 用户问题 question 一共有多少个学生只返回 SQL 语句不要返回 SQLQuery: # 5. 限制只使用 students 表 sql chain.invoke({ question: question, table_names_to_use: [students] }) print(模型原始输出) print(sql) # 6. 清理 SQLQuery: sql sql.replace(SQLQuery:, ).strip() print(\n清理后的 SQL) print(sql) # 7. 执行 SQL result db.run(sql) print(\n数据库查询结果) print(result)Memory RAG 结合版历史信息不是模型自己记住的而是程序帮它存下来并在下一次提问时重新塞回 Prompt 里。也就是说LLM 本身没有长期记忆。它每次调用时只看当前传给它的内容。模型 API 默认是无状态的所谓记忆是应用程序、框架或平台帮你保存历史并重新传给模型。第 1 轮用户问 Llama 2 是什么 系统从文档库检索 Llama 2 资料然后回答。 第 2 轮用户问“它开源吗” 用户没有说“Llama 2”但 Memory 里有上一轮聊天历史所以系统知道“它”指 Llama 2。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.chat_history import InMemoryChatMessageHistory load_dotenv() # 1. 准备大模型 llm ChatOpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, modelqwen-plus, ) # 2. 准备一个非常简单的“文档库” # 真实 RAG 里这里通常会换成 PDF 向量数据库 documents [ { title: Llama 2, content: Llama 2 是 Meta 发布的大语言模型系列包含 7B、13B、70B 等参数规模并提供聊天优化版本。 }, { title: Llama 2 开源情况, content: Llama 2 对研究和商业使用开放但需要遵守 Meta 的许可协议。 }, { title: LangChain, content: LangChain 是一个用于构建大模型应用的框架可以组合模型、提示词、工具、检索器和记忆模块。 }, ] # 3. 写一个非常简单的检索函数 # 它会根据“当前问题 历史消息”从 documents 里找相关内容 def simple_retrieve(inputs: dict) - dict: question inputs[input] chat_history inputs.get(chat_history, []) history_text for message in chat_history: history_text message.content \n search_text history_text \n question matched_docs [] for doc in documents: if Llama in search_text or llama in search_text or 它 in question: if Llama in doc[title]: matched_docs.append(doc[content]) if LangChain in search_text or langchain in search_text: if LangChain in doc[title]: matched_docs.append(doc[content]) context \n.join(matched_docs) return { input: question, chat_history: chat_history, context: context, } # 4. 准备 Prompt prompt ChatPromptTemplate.from_messages([ ( system, 你是一个文档问答助手。 你需要根据下面的文档内容回答用户问题。 文档内容 {context} 如果文档内容不足以回答就说“根据已有文档我不知道”。 回答要简洁。 ), MessagesPlaceholder(variable_namechat_history), (human, {input}), ]) # 5. 构建 RAG 链 # simple_retrieve负责检索文档 # prompt把文档、历史消息、用户问题组装成提示词 # llm生成回答 # StrOutputParser把模型输出转成字符串 rag_chain ( RunnableLambda(simple_retrieve) | prompt | llm | StrOutputParser() ) # 6. 准备聊天历史存储 # 这里用内存字典保存不是永久保存程序关闭就没了 session_store {} def get_session_history(session_id: str): if session_id not in session_store: session_store[session_id] InMemoryChatMessageHistory() return session_store[session_id] # 7. 给 RAG 链加上 Memory conversation_rag_chain RunnableWithMessageHistory( rag_chain, get_session_history, input_messages_keyinput, history_messages_keychat_history, ) # 8. 开始测试 session_id user_001 answer1 conversation_rag_chain.invoke( {input: Llama 2 是什么}, config{configurable: {session_id: session_id}}, ) print(第 1 轮回答) print(answer1) answer2 conversation_rag_chain.invoke( {input: 它开源吗}, config{configurable: {session_id: session_id}}, ) print(\n第 2 轮回答) print(answer2) answer3 conversation_rag_chain.invoke( {input: 那 LangChain 是什么}, config{configurable: {session_id: session_id}}, ) print(\n第 3 轮回答) print(answer3)