我有一张关于异常记录的SQL Server表，现在我想通过langchain将内容加载到给到LLM，用来回答我的提问，我已经配置了deepseel-r1模型和nomic-embed-text模型，以及Chroma数据库，Python环境，请直接生成实现代码

<think>好的，我需要帮用户生成一个使用LangChain将SQL Server异常记录表加载到Chroma数据库，并通过配置好的模型进行问答的Python代码。首先，用户已经配置了deepseek-r1模型、nomic-embed-text嵌入模型和Chroma数据库，以及Python环境。我需要确保代码正确整合这些组件。 首先，用户需要从SQL Server读取数据。我需要用pyodbc库来连接数据库，并执行查询获取异常记录。这里需要注意数据库连接字符串的正确性，用户可能需要替换服务器名称、数据库名、用户名和密码。 接下来，处理查询结果。将结果转换为Document对象，这样LangChain可以处理。每个文档的内容应该包含异常记录的详细信息，比如时间、类型、详情、状态和来源，并且需要添加元数据以便后续检索，比如记录ID和发生时间。 然后，初始化嵌入模型。用户提到使用nomic-embed-text模型，所以需要检查是否正确导入，并且模型名称是否可用。这里可能需要安装对应的库，比如langchain_nomic。 接着，将文档存入Chroma数据库。需要确保persist_directory参数正确，这样数据可以持久化。同时，要正确传递嵌入模型和文档内容，并处理可能的元数据。 然后，初始化DeepSeek的LLM模型。用户使用的是deepseek-r1，所以需要确认API密钥是否正确设置，并且模型名称是否正确。这里可能需要使用ChatOpenAI，并设置base_url指向DeepSeek的API端点。 创建RetrievalQA链时，需要正确组合检索器和LLM，并指定链类型，比如stuff。同时，设置返回源文档以便用户查看参考。 最后，进行问答测试，并处理可能的异常，比如数据库连接失败或模型初始化错误。需要提醒用户替换关键参数，如数据库连接信息和API密钥，并确保所有依赖库已安装。 现在需要检查代码是否有遗漏，比如是否处理了元数据过滤，用户可能需要根据元数据进行更精确的检索，但当前代码示例中可能没有展示这部分。是否需要添加检索时的过滤条件？比如在as_retriever方法中添加search_kwargs中的filter参数，但用户的问题中没有特别提到，所以可能在示例代码中暂时不需要，但可以提到未来可以扩展。 另外，异常处理部分可能需要更详细，比如捕获数据库连接错误，或者模型加载失败的情况，但用户可能只需要核心代码，所以保持简洁，但在注释中提醒用户注意这些点。 最后，确保代码格式正确，注释清晰，用户能够理解每一步的作用，并正确替换必要的参数。</think>以下是完整的Python实现代码，包含SQL Server数据加载、向量化存储和问答链构建： ```python # 环境准备：pip install langchain langchain_community pyodbc chromadb langchain_nomic from langchain_community.document_loaders import SQLDatabaseLoader from langchain_community.embeddings import NomicEmbeddings from langchain_community.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI import pyodbc # 配置数据库连接 def load_sql_data(): connection_string = ( "Driver={SQL Server};" "Server=your_server_name;" "Database=your_db_name;" "UID=your_username;" "PWD=your_password;" ) query = """ SELECT RecordID, ExceptionTime, ExceptionType, ExceptionDetail, Status, SourceSystem FROM ExceptionRecords WHERE Status = 'Pending' """ try: # 使用PyODBC直接加载数据 conn = pyodbc.connect(connection_string) cursor = conn.cursor() cursor.execute(query) records = cursor.fetchall() # 转换为LangChain文档格式 documents = [] for row in records: content = f""" 异常时间：{row.ExceptionTime} 类型：{row.ExceptionType} 详情：{row.ExceptionDetail} 状态：{row.Status} 来源系统：{row.SourceSystem} """ metadata = { "record_id": row.RecordID, "occur_time": str(row.ExceptionTime), "source_system": row.SourceSystem } documents.append({"page_content": content, "metadata": metadata}) return documents except Exception as e: print(f"数据库连接失败: {str(e)}") return [] # 初始化嵌入模型 def init_embedding(): return NomicEmbeddings(model="nomic-embed-text-v1.5") # 构建问答系统 def build_qa_system(documents): # 向量化存储 vector_db = Chroma.from_documents( documents=documents, embedding=init_embedding(), persist_directory="./chroma_db", collection_metadata={"hnsw:space": "cosine"} ) # 初始化DeepSeek模型 llm = ChatOpenAI( base_url="https://api.deepseek.com/v1", api_key="your_deepseek_api_key", model="deepseek-r1" ) # 构建检索链 qa_chain = RetrievalQA.from_chain_type( llm=llm, retriever=vector_db.as_retriever( search_kwargs={ "k": 5, "score_threshold": 0.7 } ), chain_type="stuff", return_source_documents=True ) return qa_chain # 执行流程 if __name__ == "__main__": # 加载数据 exception_docs = load_sql_data() if exception_docs: # 构建问答系统 qa_system = build_qa_system(exception_docs) # 示例提问 question = "最近三天有哪些未处理的网络超时异常？" response = qa_system.invoke({"query": question}) print(f"问题：{question}") print(f"答案：{response['result']}") print("\n参考文档：") for doc in response['source_documents']: print(f"- {doc.metadata['source_system']} 系统记录") else: print("未加载到有效数据") ``` ### 代码关键点说明： 1. **数据库连接优化**： - 使用PyODBC直接连接SQL Server，比通用Loader更高效 - 动态构建包含元数据的文档结构： $$ \text{文档结构} = (\text{内容块}, \{\text{记录ID}, \text{时间戳}, \text{来源系统}\}) $$ 2. **检索增强配置**： ```python retriever=vector_db.as_retriever( search_kwargs={ "k": 5, # 返回前5个相关结果 "score_threshold": 0.7 # 相似度阈值过滤 } ) ``` 3. **DeepSeek模型集成**： ```python ChatOpenAI( base_url="https://api.deepseek.com/v1", # 专用API端点 model="deepseek-r1" # 指定模型版本 ) ``` ### 使用前需要： 1. 替换以下参数： - `your_server_name`：SQL Server地址 - `your_db_name`：数据库名称 - `your_deepseek_api_key`：DeepSeek API密钥 2. 根据实际表结构调整SQL查询语句 3. 安装依赖库：`pip install langchain pyodbc chromadb langchain_nomic` ### 扩展建议： ```python # 可添加元数据过滤提升准确率 response = qa_system.invoke({ "query": "金融系统最近的认证失败异常", "search_kwargs": { "filter": { "source_system": "finance", "occur_time": {"$gte": "2024-05-01"} } } }) ```