活动介绍
file-type

myCBR Python实现字符串相似度检测方法

版权申诉
5星 · 超过95%的资源 | 51KB | 更新于2024-12-04 | 135 浏览量 | 1 下载量 举报 收藏
download 限时特惠:#19.90
在这个文件标题中,我们可以看到几个关键的IT知识点,分别是“myCBR”,“外部相似度方法”,“Python编程语言”,以及“字符串相似性”。下面将详细解释这些概念和知识点。 首先,"myCBR"是指一个案例基础推理(Case-Based Reasoning,简称CBR)系统。案例基础推理是一种人工智能技术,用于解决新问题,通过寻找与新问题相似的旧问题(案例)来得到解决方案。myCBR是一种开源的CBR平台,它允许用户创建CBR应用程序,支持多种编程语言,包括Python。CBR系统在各种领域中都有应用,如客户服务、医疗诊断、电子商务等。 接着,“外部相似度方法”是案例基础推理系统中一个重要的概念。在CBR系统中,为了找到与新问题(查询)相似的旧问题(案例),需要一种方法来计算案例之间的相似度。外部相似度方法是相对于内部相似度方法而言的。内部方法基于案例的属性值之间的直接比较,而外部方法则是基于额外的度量标准或算法来计算相似度,例如通过自然语言处理技术来分析文本数据的相似性。在这个特定的代码片段中,外部相似度方法是根据字符串中特定字符(字母“e”)的数量来判断两个字符串的相似性。 标题中的"Python"是目前非常流行的一种高级编程语言,广泛用于网站开发、自动化脚本编写、数据分析、人工智能等领域。Python以其简洁易读的语法和强大的库支持而闻名,使得开发者可以快速地实现复杂的算法和应用程序。 至于“字符串相似性”,这通常是指通过特定的算法或函数来比较两个或多个字符串的相似程度。在信息检索、文本挖掘、生物信息学等领域中,字符串相似性是一个核心概念。有许多算法和技术可用于计算字符串相似性,比如编辑距离(Levenshtein距离)、余弦相似性、Jaccard相似性等。在这个案例中,字符串相似性算法简化为计算字符串中相同字符(“e”)的数量,这实际上是一种非常简单的相似性度量方式。 【压缩包子文件的文件名称列表】中的"sim"可能是代码文件或者文档的名称,由于信息有限,我们无法确定具体是什么类型的文件,但是可以推断它与相似性计算相关。 总结来说,这个文件标题和描述涉及到的内容包括: - myCBR案例基础推理系统的介绍和应用 - 外部相似度方法与内部相似度方法的区别及其在CBR中的作用 - Python编程语言在CBR系统开发中的应用 - 字符串相似性的基本概念以及一种特定的简单实现方法 这个文件可能包含了一个Python实现的示例代码,该代码利用myCBR框架来计算查询字符串和案例字符串之间基于特定字符(“e”)数量的相似性。了解这些知识点,可以帮助IT专业人士更好地理解相似性计算的原理,并在实际工作中设计和实现相应的算法。

相关推荐

filetype

import os import requests import hnswlib import numpy as np from typing import List, Tuple import re import json #TODO: 添加prompt,确保回答被{}包裹 #TODO:加分项,修改prompt,思考怎么样LLM更容易正确回答 naive_prompt = """ ----Role---- You are a helpful language model (LLM) tasked with using texts and any necessary common knowledge to answer a given question. """ class LocalSentenceSplitter: def __init__(self, chunk_size, chunk_overlap) -> None: self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap @staticmethod def _zng(paragraph): pattern = u'([!?。…!?.\n])' return [sent for sent in re.split(pattern, paragraph, flags=re.U)] @staticmethod def _zng_new(paragraph): sentence = [] #TODO: 添加split方法 # 使用自己的方法来分割句子 return sentence def split_text(self, segment): chunks, chunk_now, size_now = [], [], 0 no_left = False for s in LocalSentenceSplitter._zng_new(segment): #TODO: 添加chunk_now的判断,如果chunk_now的长度大于chunk_size,则进行分割 # 额外需要确保正确overlap return chunks def _get_overlap(self, chunk): rchunk = chunk[:] rchunk.reverse() size_now, overlap = 0, [] for s in rchunk[:-1]: overlap.append(s) size_now += len(s) if size_now > self.chunk_overlap: break overlap.reverse() return overlap, size_now class OllamaEmbedding: """ 调用本地 Ollama 的 /embeddings 接口来获取文本的Embedding。 假设 Ollama 版本支持 embeddings 功能,且在 localhost:11434 端口提供服务。 你可以根据需要修改 URL、超时、模型名称等。 """ def __init__(self, host="http://localhost:11434", model="bge-large", work_dir="work_dir"): self.host = host self.model = model self.work_dir = work_dir # 日志文件,可在此记录一些embedding的调用信息 self.log_path = os.path.join(self.work_dir, "logs.txt") os.makedirs(self.work_dir, exist_ok=True) def _log(self, msg: str): with open(self.log_path, "a", encoding="utf-8") as f: f.write(msg + "\n") def encode(self, texts: List[str]) -> np.ndarray: """ 输入若干文本,返回它们的向量(shape=[len(texts), dim])。 """ embeddings = [] for text in texts: try: payload = { "model": self.model, "prompt": text } # Ollama 的 /embeddings 接口(注意版本不同,接口可能略有差异) url = f"{self.host}/api/embeddings" r = requests.post(url, json=payload, timeout=60) r.raise_for_status() data = r.json() # data 应该包含 "embedding" 字段,这里做个简单取值 emb = data.get("embedding", []) embeddings.append(emb) self._log(f"[encode] success for text: {text[:30]}..., embedding dim={len(emb)}") except Exception as e: # 异常时可记录日志,返回一个零向量或者直接抛出 self._log(f"[encode] error for text: {text[:30]}..., error={e}") # 这里简单用全0向量顶替 embeddings.append([0.0]*1024) return np.array(embeddings) class HNSWVectorStore: """ 用 hnswlib 来管理向量索引。采用 'cosine' 作为距离度量。 """ def __init__(self, embedding_dim: int, work_dir: str = "work_dir"): self.embedding_dim = embedding_dim self.work_dir = work_dir os.makedirs(self.work_dir, exist_ok=True) # index.bin是hnswlib的索引文件,labels.npy是文本的label self.index_path = os.path.join(self.work_dir, "index.bin") self.labels_path = os.path.join(self.work_dir, "labels.npy") # 判断索引文件是否已经存在,决定是加载还是新建 if os.path.exists(self.index_path) and os.path.exists(self.labels_path): #TODO: 添加加载方法 print("[HNSWVectorStore] index loaded from disk.") else: #TODO: 添加初始化方法 print("[HNSWVectorStore] created new empty index.") def add_texts(self, texts: List[str], embeddings: np.ndarray): """ 将文本和其向量插入 HNSW 索引,并记录 labels(即文本内容)。 """ num_new = embeddings.shape[0] start_id = len(self.labels) ids = np.arange(start_id, start_id + num_new) # 插入向量 self.index.add_items(embeddings, ids) # 记录文本,以便检索时能得到对应的文本 self.labels.extend(texts) def save(self): """ 将索引和对应 labels 保存到磁盘。 """ self.index.save_index(self.index_path) np.save(self.labels_path, np.array(self.labels, dtype=object), allow_pickle=True) print("[HNSWVectorStore] index saved.") def similarity_search(self, query_embedding: np.ndarray, top_k: int = 3) -> List[str]: """ 基于向量检索最相似的 Top-K 文本。 """ # hnswlib 返回的是 (ids, distances) ids, distances = self.index.knn_query(query_embedding, k=top_k) # ids shape: (1, top_k) distances shape: (1, top_k) # 这里假设只有一个查询向量,所以取 ids[0] print(ids) result_texts = [self.labels[idx] for idx in ids[0]] return result_texts class NaiveRAG: """ 一个简易的RAG示例:用本地 Ollama 生成Embedding,再用HNSW检索相似文本,最后拼接回答。 """ def __init__(self, embedding_dim: int = 1024, work_dir="work_dir", model="bge-large"): self.work_dir = work_dir self.embedding_model = OllamaEmbedding( host="http://localhost:11434", model=model, work_dir=work_dir ) self.vector_store = HNSWVectorStore(embedding_dim=embedding_dim, work_dir=work_dir) def build_index(self, text_chunks: List[str]): """ 对text_chunks做embedding并插入索引,然后保存索引到磁盘。 """ embeddings = self.embedding_model.encode(text_chunks) # [N, dim] self.vector_store.add_texts(text_chunks, embeddings) self.vector_store.save() def generate_prompt(self, query: str, top_k: int = 3) -> str: """ 对用户query做embedding,然后从索引中检索top_k文本块,用简单方式拼接。 """ query_emb = self.embedding_model.encode([query])[0] # shape=[dim] retrieved_texts = self.vector_store.similarity_search(query_emb, top_k=top_k) #TODO:加分项添加reranker,对retrieved_texts进行rerank #print(retrieved_texts) # 这里仅用一个很简单的拼接回答示例 context = "\n".join(retrieved_texts) #answer = ( # f"You are a helpful assistant please answer the question {query} with given context:\n{context}" #) prompt = naive_prompt+f"Q:{query}\ntexts:{context}\nA:" return prompt 请修改以上代码,完成撰写 LocalSentenceSplitter 对语料进行分段,调整LocalSentenceSplitter探究分块机制对问答效果影响;利用 Embedding模型 ,完善 HNSWVectorStore构建向量索引,

filetype

--------------------------------------------------------------------------- ValueError Traceback (most recent call last) Cell In[146], line 76 73 return all_client_df 75 # 计算每日持仓余弦相似性 ---> 76 daily_positions_cos = calculate_positions_consine_similarity(multi_cpty_df[(multi_cpty_df["client_id"] == "ABRH400")], interval=90) 78 # plot_pos_sim(daily_positions_cos["date"], daily_positions_cos["pos_sim"]) 80 daily_positions_cos Cell In[146], line 68, in calculate_positions_consine_similarity(df, interval) 66 similarity_median = calculate_cosine_similarity(merged, interval) 67 print(similarity_median) ---> 68 similarity_results = pd.DataFrame({"client_id": client, "pos_sim": similarity_median}) 70 all_client_result.append(similarity_results) 72 all_client_df = pd.concat(all_client_result, axis=1) File c:\Users\matianht\.conda\envs\nomura\lib\site-packages\pandas\core\frame.py:778, in DataFrame.__init__(self, data, index, columns, dtype, copy) 772 mgr = self._init_mgr( 773 data, axes={"index": index, "columns": columns}, dtype=dtype, copy=copy 774 ) 776 elif isinstance(data, dict): 777 # GH#38939 de facto copy defaults to False only in non-dict cases --> 778 mgr = dict_to_mgr(data, index, columns, dtype=dtype, copy=copy, typ=manager) 779 elif isinstance(data, ma.MaskedArray): 780 from numpy.ma import mrecords File c:\Users\matianht\.conda\envs\nomura\lib\site-packages\pandas\core\internals\construction.py:503, in dict_to_mgr(data, index, columns, dtype, typ, copy) 499 else: 500 # dtype check to exclude e.g. range objects, scalars 501 arrays = [x.copy() if hasattr(x, "dtype") else x for x in arrays] --> 503 return arrays_to_mgr(arrays, columns, index, dtype=dtype, typ=typ, consolidate=copy) File c:\Users\matianht\.conda\envs\nomura\lib\site-packages\pandas\core\internals\construction.py:114, in arrays_to_mgr(arrays, columns, index, dtype, verify_integrity, typ, consolidate) 111 if verify_integrity: 112 # figure out the index, if necessary 113 if index is None: --> 114 index = _extract_index(arrays) 115 else: 116 index = ensure_index(index) File c:\Users\matianht\.conda\envs\nomura\lib\site-packages\pandas\core\internals\construction.py:667, in _extract_index(data) 664 raise ValueError("Per-column arrays must each be 1-dimensional") 666 if not indexes and not raw_lengths: --> 667 raise ValueError("If using all scalar values, you must pass an index") 669 if have_series: 670 index = union_indexes(indexes) ValueError: If using all scalar values, you must pass an index 这是什么意思

钱亚锋
  • 粉丝: 124
上传资源 快速赚钱