一个案例演示什么是RAG+Prompt工程

# 导入必要的库和模块
import os  # 操作系统接口模块
from openai import OpenAI  # OpenAI客户端库
from langchain.document_loaders import TextLoader  # 文本加载器
from langchain.text_splitter import CharacterTextSplitter  # 文本分割器
from langchain.embeddings import HuggingFaceEmbeddings  # HuggingFace嵌入模型
from langchain.vectorstores import FAISS  # Facebook的向量数据库

# 初始化DashScope客户端
def init_client():
    # 创建并返回OpenAI客户端实例
    return OpenAI(
        api_key="xxxxxxxxxxxxxxxxxx",  # 替换为你的API Key，用于身份验证
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # API基础URL
    )

# 文档处理函数
def process_documents(file_path):
    try:
        # 创建文本加载器实例，加载指定路径的文件
        loader = TextLoader(file_path)
        # 加载文档内容
        documents = loader.load()
        
        # 创建文本分割器实例，配置分割参数
        text_splitter = CharacterTextSplitter(
            chunk_size=1000,  # 每个文本块的最大字符数
            chunk_overlap=200,  # 块之间的重叠字符数
            separator="\n"  # 按换行符分割
        )
        # 对文档进行分割并返回结果
        return text_splitter.split_documents(documents)
    except Exception as e:
        # 如果出现异常，打印错误信息并返回None
        print(f"文档处理错误: {str(e)}")
        return None

# 创建向量存储函数
def create_vector_store(documents):
    try:
        # 初始化HuggingFace嵌入模型
        embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
        # 使用FAISS从文档创建向量存储
        return FAISS.from_documents(documents, embeddings)
    except Exception as e:
        # 如果出现异常，打印错误信息并返回None
        print(f"创建向量存储错误: {str(e)}")
        return None

# RAG问答函数
def rag_query(client, vector_store, query):
    try:
        # 检索相关文档：在向量库中搜索与查询最相似的3个文档片段
        docs = vector_store.similarity_search(query, k=3)
        # 将检索到的文档内容合并为一个字符串，用换行符连接
        context = "\n".join([doc.page_content for doc in docs])
        
        # 构建提示词(messages)列表，用于与AI模型交互
        messages = [
            # 系统消息：设定AI的角色和行为
            {'role': 'system', 'content': '你是一个基于上下文算命大师，你可以根据用户的出生日期进行精准的算命，口吻上面要像一个真人，回答要有神棍一样的话术。'},
            # 用户消息：包含检索到的上下文和用户问题
            {'role': 'user', 'content': f"上下文：\n{context}\n\n问题：{query}"}
        ]
        
        # 调用DashScope API生成回答
        response = client.chat.completions.create(
            model="qwen-plus",  # 指定使用的模型
            messages=messages,  # 传入构建好的消息列表
            temperature=0.7  # 控制生成随机性的参数(0-1)
        )
        # 返回模型生成的第一个回答内容
        return response.choices[0].message.content
    except Exception as e:
        # 如果出现异常，返回错误信息
        return f"查询过程中出错: {str(e)}"

# 主函数
def main():
    # 初始化OpenAI客户端
    client = init_client()
    
    # 处理文档
    print("正在加载和处理文档...")
    # 加载并处理指定路径的文档
    documents = process_documents("/Users/xxxxxxxxx/Desktop/documents.txt")
    # 如果文档处理失败，直接返回
    if not documents:
        return
    
    # 创建向量存储
    print("正在创建向量数据库...")
    # 从处理好的文档创建向量数据库
    vector_store = create_vector_store(documents)
    # 如果向量库创建失败，直接返回
    if not vector_store:
        return
    
    # 交互式问答循环
    print("\nRAG系统已准备就绪，输入'退出'结束对话")
    while True:
        # 获取用户输入的问题，去除首尾空格
        query = input("\n请输入问题：").strip()
        # 检查是否为退出命令
        if query.lower() in ['退出', 'exit', 'quit']:
            break  # 退出循环
            
        # 调用RAG查询函数获取回答
        response = rag_query(client, vector_store, query)
        # 打印AI的回答
        print("\n回答：", response)

# Python脚本入口
if __name__ == "__main__":
    # 当脚本直接运行时，调用主函数
    main()