asp.net 网站开发的技术优势网络企业

asp.net 网站开发的技术优势,网络企业,网页设计图片轮播,wordpress 非小工具形式 微博秀之前探索了如何用DSPy优化RAG prompt。 https://blog.csdn.net/liliang199/article/details/155893634 这里首先设定chromadb向量检索系统#xff0c;然后通过DSPy优化基于retriever的RAG检索功能。 所用测试例和代码修改自网络资料。 1 retriever设置 1.1 向量计算设置 这…之前探索了如何用DSPy优化RAG prompt。https://blog.csdn.net/liliang199/article/details/155893634这里首先设定chromadb向量检索系统然后通过DSPy优化基于retriever的RAG检索功能。所用测试例和代码修改自网络资料。1 retriever设置1.1 向量计算设置这里基于sentence-transformer框架并采用bge-m3向量模型。MiniLM是微软开发的轻量级语言模型以较小参数量和计算成本实现。all-MiniLM-L6-v2属于 MiniLM 系列通过知识蒸馏从更大模型压缩而来旨在保持较高性能同时减少计算资源需求。代码示例如下import osos.environ[HF_ENDPOINT] https://hf-mirror.comfrom sentence_transformers import SentenceTransformerembedding_model SentenceTransformer(all-MiniLM-L6-v2)这里设置HF_ENDPOINT是因为hf访问受限。1.2 向量库设置chromadb的目标是帮助用户更加便捷地构建大模型应用更加轻松的将知识、事实和技能等现实世界中的文档整合进大模型中。这里选用chromadb向量库安装过程参考如下链接https://blog.csdn.net/liliang199/article/details/149542926设置示例如下所示向量检索使用余弦相似度。persist_directory ./curr_chromadbcollection_name “knowledge_test”client chromadb.PersistentClient(persist_directory)collection client.get_or_create_collection(namecollection_name,metadata{hnsw:space: cosine} # 使用余弦相似度)1.3 向量检索示例以下是应用sentence-transformer和chromadb的向量检索器类可修改后应用于生产环境。包括向量库初始化文档划分、向量生成、向量检索等检索器要求的基本功能。1retrieverimport os os.environ[HF_ENDPOINT] https://hf-mirror.com import dspy import chromadb from chromadb.config import Settings from typing import List, Dict, Any, Optional import hashlib import logging from sentence_transformers import SentenceTransformer import numpy as np # 设置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class VectorRetriever: 完整的向量检索器支持文档初始化、嵌入生成和语义检索。 使用ChromaDB作为向量存储SentenceTransformer生成嵌入。 def __init__( self, collection_name: str knowledge_base, embedding_model: str BAAI/bge-m3, # 轻量且效果不错的模型 persist_directory: str ./chroma_db, chunk_size: int 500, chunk_overlap: int 50 ): 初始化检索器 Args: collection_name: 集合名称 embedding_model: 嵌入模型名称 persist_directory: 向量数据库存储路径 chunk_size: 文本分块大小字符数 chunk_overlap: 分块重叠大小字符数 self.embedding_model SentenceTransformer(embedding_model) self.chunk_size chunk_size self.chunk_overlap chunk_overlap # 初始化ChromaDB客户端 self.client chromadb.PersistentClient(persist_directory) # chromadb.Client(Settings( # chroma_db_implduckdbparquet, # persist_directorypersist_directory, # anonymized_telemetryFalse # 禁用遥测 # )) # 获取或创建集合 self.collection self.client.get_or_create_collection( namecollection_name, metadata{hnsw:space: cosine} # 使用余弦相似度 ) logger.info(f检索器初始化完成使用模型: {embedding_model}) def _generate_chunk_id(self, text: str, source: str, chunk_index: int) - str: 生成唯一的块ID content f{source}_{chunk_index}_{text[:50]} return hashlib.md5(content.encode()).hexdigest()[:16] def split_text(self, text: str) - List[str]: 将长文本分割为重叠的块 Args: text: 输入文本 Returns: 文本块列表 if len(text) self.chunk_size: return [text] chunks [] start 0 while start len(text): # 计算块结束位置 end start self.chunk_size # 如果不在文本末尾尝试在句号、逗号或空格处分割 if end len(text): # 查找合适的分割点 for split_point in range(end, start self.chunk_size - 100, -1): if split_point len(text) and text[split_point] in .。!?,;\n\t : end split_point 1 # 包含分割字符 break chunk text[start:end].strip() if chunk: # 忽略空块 chunks.append(chunk) # 移动起始位置考虑重叠 start end - self.chunk_overlap # 防止无限循环 if start len(text) or (start end and end len(text)): break logger.info(f将文本分割为 {len(chunks)} 个块) return chunks def prepare_documents( self, documents: List[Dict[str, Any]], batch_size: int 32 ) - List[Dict[str, Any]]: 准备文档以供索引 Args: documents: 文档列表每个文档是字典必须包含content字段 batch_size: 批处理大小 Returns: 处理后的文档块列表 all_chunks [] total_docs len(documents) logger.info(f开始处理 {total_docs} 个文档...) for doc_idx, doc in enumerate(documents, 1): content doc.get(content, ) if not content: logger.warning(f文档 {doc_idx} 无内容跳过) continue # 获取元数据排除content字段 metadata {k: v for k, v in doc.items() if k ! content} metadata[source_doc_id] doc.get(id, fdoc_{doc_idx}) # 分割文本 chunks self.split_text(content) # 为每个块创建记录 for chunk_idx, chunk in enumerate(chunks): chunk_id self._generate_chunk_id(chunk, metadata[source_doc_id], chunk_idx) chunk_metadata metadata.copy() chunk_metadata.update({ chunk_id: chunk_idx, total_chunks: len(chunks), char_length: len(chunk), doc_index: doc_idx - 1 }) all_chunks.append({ id: chunk_id, text: chunk, metadata: chunk_metadata, embedding: None # 稍后生成 }) if doc_idx % 10 0 or doc_idx total_docs: logger.info(f已处理 {doc_idx}/{total_docs} 个文档生成 {len(all_chunks)} 个块) logger.info(f文档准备完成共生成 {len(all_chunks)} 个文本块) return all_chunks def generate_embeddings( self, chunks: List[Dict[str, Any]], batch_size: int 32 ) - List[Dict[str, Any]]: 为文本块生成嵌入向量 Args: chunks: 文本块列表 batch_size: 批处理大小 Returns: 包含嵌入向量的文本块列表 if not chunks: return chunks logger.info(f开始为 {len(chunks)} 个文本块生成嵌入...) # 提取文本 texts [chunk[text] for chunk in chunks] # 分批生成嵌入 embeddings [] for i in range(0, len(texts), batch_size): batch_texts texts[i:i batch_size] batch_embeddings self.embedding_model.encode( batch_texts, show_progress_barFalse, normalize_embeddingsTrue # 归一化以使用余弦相似度 ) embeddings.extend(batch_embeddings) if (i // batch_size) % 10 0: logger.info(f已生成 {min(i batch_size, len(texts))}/{len(texts)} 个嵌入) # 将嵌入添加到块中 for chunk, embedding in zip(chunks, embeddings): chunk[embedding] embedding.tolist() if hasattr(embedding, tolist) else embedding logger.info(嵌入生成完成) return chunks def index_documents( self, documents: List[Dict[str, Any]], clear_existing: bool False ) - int: 索引文档到向量数据库 Args: documents: 原始文档列表 clear_existing: 是否清除现有索引 Returns: 索引的文档块数量 if clear_existing: self.client.delete_collection(self.collection.name) self.collection self.client.get_or_create_collection( nameself.collection.name, metadata{hnsw:space: cosine} ) logger.info(已清除现有索引) # 准备文档 chunks self.prepare_documents(documents) if not chunks: logger.warning(没有可索引的文档块) return 0 # 生成嵌入 chunks_with_embeddings self.generate_embeddings(chunks) # 准备数据用于添加到集合 ids [chunk[id] for chunk in chunks_with_embeddings] texts [chunk[text] for chunk in chunks_with_embeddings] metadatas [chunk[metadata] for chunk in chunks_with_embeddings] embeddings [chunk[embedding] for chunk in chunks_with_embeddings] # 分批添加到集合 batch_size 100 total_added 0 for i in range(0, len(ids), batch_size): batch_ids ids[i:i batch_size] batch_texts texts[i:i batch_size] batch_metadatas metadatas[i:i batch_size] batch_embeddings embeddings[i:i batch_size] self.collection.add( idsbatch_ids, documentsbatch_texts, metadatasbatch_metadatas, embeddingsbatch_embeddings ) total_added len(batch_ids) logger.info(f已索引 {total_added}/{len(ids)} 个文档块) logger.info(f文档索引完成共索引 {total_added} 个文档块) return total_added def retrieve( self, query: str, k: int 3, score_threshold: float 0.5, include_metadata: bool True ) - List[Any]: 检索与查询最相关的文档 Args: query: 查询文本 k: 返回结果数量 score_threshold: 相似度阈值 include_metadata: 是否包含元数据 Returns: 检索结果列表 if self.collection.count() 0: logger.warning(集合为空请先索引文档) return [] # 生成查询嵌入 query_embedding self.embedding_model.encode( query, normalize_embeddingsTrue ).tolist() # 执行查询 results self.collection.query( query_embeddings[query_embedding], n_resultsmin(k * 2, 20), # 获取更多结果用于过滤 include[documents, metadatas, distances] ) # 处理结果 retrieved_docs [] if results[documents] and results[documents][0]: for i, (doc, distance, metadata) in enumerate(zip( results[documents][0], results[distances][0], results[metadatas][0] if results[metadatas] else [{}] * len(results[documents][0]) )): # 将距离转换为相似度分数余弦相似度 similarity 1 - distance if similarity score_threshold: result_item { text: doc, score: similarity, metadata: metadata if include_metadata else None } retrieved_docs.append(result_item) # 达到所需数量时停止 if len(retrieved_docs) k: break # 按分数排序 retrieved_docs.sort(keylambda x: x[score], reverseTrue) logger.info(f检索到 {len(retrieved_docs)} 个相关文档 (阈值: {score_threshold})) return retrieved_docs def get_collection_stats(self) - Dict[str, Any]: 获取集合统计信息 count self.collection.count() # 获取一些样本的元数据以了解结构 sample self.collection.peek(limit1) metadata_keys set() if sample[metadatas] and sample[metadatas][0]: metadata_keys set(sample[metadatas][0].keys()) return { total_chunks: count, metadata_fields: list(metadata_keys), collection_name: self.collection.name }这里尝试初始化和测试检索器示例代码如下所示。2retriever testretriever的测试代码如下所示包括准备示例文档、初始化检索器、索引文档、测试检索等过程。# 使用示例 def retriever_test(): 使用VectorRetriever的完整示例 # 1. 准备示例文档 sample_documents [ { id: doc_1, title: 人工智能简介, content: 人工智能AI是计算机科学的一个分支致力于创建能够执行通常需要人类智能的任务的系统。 这些任务包括视觉感知、语音识别、决策和语言翻译。AI可以分为两类弱人工智能和强人工智能。 弱人工智能专用于特定任务而强人工智能具有通用的人类认知能力。 机器学习是AI的一个子集它使计算机能够在没有明确编程的情况下从数据中学习。 深度学习是机器学习的一个子领域使用神经网络模拟人脑的工作方式。 , category: 科技, author: 张三, year: 2023 }, { id: doc_2, title: 机器学习算法, content: 机器学习算法可以分为三大类监督学习、无监督学习和强化学习。 监督学习使用标记数据训练模型例如分类和回归任务。常见算法包括线性回归、逻辑回归、支持向量机和神经网络。 无监督学习处理未标记数据旨在发现数据中的模式。聚类和降维是无监督学习的典型任务。K-means和PCA是常用算法。 强化学习涉及智能体通过与环境交互学习最优策略。它通过奖励和惩罚机制进行学习常用于游戏和机器人控制。 , category: 科技, author: 李四, year: 2022 }, { id: doc_3, title: 自然语言处理, content: 自然语言处理NLP是AI的一个分支专注于计算机与人类语言之间的交互。 NLP任务包括文本分类、情感分析、命名实体识别、机器翻译和问答系统。 近年来基于Transformer的模型如BERT、GPT彻底改变了NLP领域。 这些模型使用自注意力机制能够更好地理解上下文和语言语义。 NLP的应用包括聊天机器人、搜索引擎、语音助手和自动摘要等。 , category: 科技, author: 王五, year: 2024 } ] # 2. 初始化检索器 print(初始化向量检索器...) retriever VectorRetriever( collection_nameai_knowledge_base, embedding_modelall-MiniLM-L6-v2, persist_directory./chroma_test_ai, chunk_size400, chunk_overlap30 ) # 3. 索引文档 print(\n索引文档...) indexed_count retriever.index_documents( sample_documents, clear_existingTrue # 首次运行设为True后续可以设为False以增量添加 ) print(f已索引 {indexed_count} 个文档块) # 4. 查看统计信息 stats retriever.get_collection_stats() print(f\n集合统计:) print(f 文档块总数: {stats[total_chunks]}) print(f 元数据字段: {stats[metadata_fields]}) # 5. 测试检索 test_queries [ 什么是机器学习, 监督学习和无监督学习有什么区别, 介绍一下自然语言处理 ] print(\n测试检索功能:) for query in test_queries: print(f\n查询: {query}) results retriever.retrieve(query, k2, score_threshold0.3) for i, result in enumerate(results, 1): print(f 结果 {i} (分数: {result[score]:.3f}):) print(f 文本: {result[text][:100]}...) if result[metadata]: print(f 来源: {result[metadata].get(title, 未知)}) return retriever retriever retriever_test()输出如下INFO:sentence_transformers.SentenceTransformer:Use pytorch device_name: cpuINFO:sentence_transformers.SentenceTransformer:Load pretrained SentenceTransformer: all-MiniLM-L6-v2初始化向量检索器...INFO:chromadb.telemetry.product.posthog:Anonymized telemetry enabled. See https://docs.trychroma.com/telemetry for more information.INFO:__main__:检索器初始化完成使用模型: all-MiniLM-L6-v2INFO:__main__:已清除现有索引INFO:__main__:开始处理 3 个文档...INFO:__main__:已处理 3/3 个文档生成 3 个块INFO:__main__:文档准备完成共生成 3 个文本块INFO:__main__:开始为 3 个文本块生成嵌入...INFO:__main__:已生成 3/3 个嵌入INFO:__main__:嵌入生成完成INFO:__main__:已索引 3/3 个文档块INFO:__main__:文档索引完成共索引 3 个文档块索引文档...已索引 3 个文档块集合统计:文档块总数: 3元数据字段: [chunk_id, category, id, total_chunks, char_length, year, source_doc_id, author, title, doc_index]测试检索功能:查询: 什么是机器学习Batches: 100%|██████████| 1/1 [00:0000:00, 87.86it/s]INFO:__main__:检索到 2 个相关文档 (阈值: 0.3)结果 1 (分数: 0.475):文本:机器学习算法可以分为三大类监督学习、无监督学习和强化学习。监督学习使用标记数据训练模型例如分类和回归任务。常见算法包括线性回归、逻辑回归、支持...来源: 机器学习算法结果 2 (分数: 0.375):文本:人工智能AI是计算机科学的一个分支致力于创建能够执行通常需要人类智能的任务的系统。这些任务包括视觉感知、语音识别、决策和语言翻译。AI可以分为...来源: 人工智能简介查询: 监督学习和无监督学习有什么区别Batches: 100%|██████████| 1/1 [00:0000:00, 126.13it/s]INFO:__main__:检索到 2 个相关文档 (阈值: 0.3)结果 1 (分数: 0.478):文本:机器学习算法可以分为三大类监督学习、无监督学习和强化学习。监督学习使用标记数据训练模型例如分类和回归任务。常见算法包括线性回归、逻辑回归、支持...来源: 机器学习算法结果 2 (分数: 0.416):文本:自然语言处理NLP是AI的一个分支专注于计算机与人类语言之间的交互。NLP任务包括文本分类、情感分析、命名实体识别、机器翻译和问答系统。...来源: 自然语言处理查询: 介绍一下自然语言处理Batches: 100%|██████████| 1/1 [00:0000:00, 141.48it/s]INFO:__main__:检索到 2 个相关文档 (阈值: 0.3)结果 1 (分数: 0.330):文本:自然语言处理NLP是AI的一个分支专注于计算机与人类语言之间的交互。NLP任务包括文本分类、情感分析、命名实体识别、机器翻译和问答系统。...来源: 自然语言处理结果 2 (分数: 0.318):文本:人工智能AI是计算机科学的一个分支致力于创建能够执行通常需要人类智能的任务的系统。这些任务包括视觉感知、语音识别、决策和语言翻译。AI可以分为...来源: 人工智能简介2 RAG设置2.1 配置DSPy签名沿用之前应用习惯这里继续使用ollama/gemma3n:e2b模型并假设ollama环境可用。在此基础上设置:context: 问题相关的上下文背景通过retriever向量检索获得question: 用户查询问题answer: llm依据context和question生成简洁、准确的答案。示例代码如下。# 配置DSPy # lm dspy.LM(openai/gpt-4o-mini) lm dspy.LM(modelollama/gemma3n:e2b, api_basehttp://localhost:11434) dspy.configure(lmlm) # 定义RAG签名 class RAGSignature(dspy.Signature): 基于上下文回答问题 context dspy.InputField(desc相关背景信息) question dspy.InputField() answer dspy.OutputField(desc简洁、准确的答案)2.2 配置RAG系统这里进一步设置RAG系统在生成环节应用DSPy的ChainOfThought通过思维链进行优化。检索具体过程为:首先retriever检索question相关的上下文并合并然后基于检索上下文和问题生成答案这一步会应用DSPy优化后的prompt。最后将上下文、问题、答案返回给用户。示例代码如下。# 创建RAG模块 class EnhancedRAG(dspy.Module): def __init__(self, retriever): super().__init__() self.retriever retriever self.generate_answer dspy.ChainOfThought(RAGSignature) def forward(self, question): # 检索相关文档 retrieved self.retriever.retrieve(question, k3) if not retrieved: return dspy.Prediction( context, answer未找到相关信息, sources[] ) # 合并上下文 contexts [r[text] for r in retrieved] context_str \n\n.join(contexts) # 生成答案 prediction self.generate_answer( contextcontext_str, questionquestion ) # 返回结果 return dspy.Prediction( contextcontext_str, answerprediction.answer, sources[r[metadata] for r in retrieved if r[metadata]], reasoninggetattr(prediction, reasoning, ) ) # 创建RAG系统并测试 rag_system EnhancedRAG(retriever)2.3 测试RAG系统这里测试刚才定义好的RAG系统直接输入用户问题。test_question 机器学习有哪些主要类型测试代码示例如下test_question 机器学习有哪些主要类型 print(f\nRAG系统测试问题: {test_question}) result rag_system(test_question) print(f答案: {result.answer}) print(f来源文档数: {len(result.sources)})输出如下RAG检索到test_question相关的问题并应用LLM进行回答。并且回答内容简洁、准确符合DSPy签名中定义的基本需求。RAG系统测试问题: 机器学习有哪些主要类型Batches: 100%|██████████| 1/1 [00:0000:00, 84.42it/s]INFO:__main__:检索到 1 个相关文档 (阈值: 0.5)答案: 监督学习、无监督学习和强化学习来源文档数: 1reference---如何用DSPy优化RAG prompt示例https://blog.csdn.net/liliang199/article/details/155893634开源向量LLM - BGE (BAAI General Embedding)https://blog.csdn.net/liliang199/article/details/149773775mac测试ollama llamaindexhttps://blog.csdn.net/liliang199/article/details/149542926chromedbhttps://docs.trychroma.com/docs/overview/migrationbge embeddinghttps://github.com/FlagOpen/FlagEmbedding一篇吃透模型all-MiniLM-L6-v2https://zhuanlan.zhihu.com/p/27730652617向量数据库Chroma极简教程https://zhuanlan.zhihu.com/p/665715823