Cogito 3B开源模型教程：使用Ollama Embeddings构建语义检索模块-编程阁

Cogito 3B开源模型教程：使用Ollama Embeddings构建语义检索模块

1. 认识Cogito 3B开源模型

Cogito v1预览版是Deep Cogito推出的混合推理模型系列，这个3B版本在大多数标准基准测试中都表现出色，超越了同等规模的其他开源模型。与LLaMA、DeepSeek和Qwen等知名模型的同类版本相比，Cogito 3B展现出了更强的能力。

Cogito模型是经过指令调优的生成模型，采用文本输入和文本输出的方式。最重要的是，所有模型都以开放许可发布，允许商业使用，这为开发者提供了很大的便利。

这个模型的独特之处在于它是混合推理模型。它既可以像标准大语言模型一样直接回答问题，也可以在回答前进行自我反思，类似于推理模型的工作方式。这种双重能力让它在处理复杂问题时更加得心应手。

模型使用迭代蒸馏和放大（IDA）策略进行训练，这是一种通过迭代自我改进来实现智能提升的高效方法。模型特别针对编码、STEM学科、指令执行和通用帮助性进行了优化，在同等规模的模型中具有显著更高的多语言支持、编码能力和工具调用能力。

每个模型都支持超过30种语言，并且拥有128k的超长上下文长度，这意味着它可以处理很长的文档和复杂的对话。

2. 环境准备与Ollama部署

2.1 安装Ollama框架

首先我们需要安装Ollama框架，这是一个专门用于本地运行大语言模型的工具。根据你的操作系统选择相应的安装方式：

# 在Linux/macOS上安装 curl -fsSL https://ollama.ai/install.sh | sh # 在Windows上可以通过WSL安装，或者下载官方安装包

安装完成后，验证Ollama是否正常工作：

ollama --version

2.2 拉取Cogito 3B模型

通过Ollama拉取Cogito 3B模型非常简单：

ollama pull cogito:3b

这个过程可能会花费一些时间，取决于你的网络速度。模型大小约为3B参数，下载量在几个GB左右。

2.3 验证模型运行

拉取完成后，测试模型是否正常运行：

ollama run cogito:3b

如果看到模型提示符，说明安装成功。你可以尝试输入一些简单问题来测试模型响应。

3. 构建语义检索模块

3.1 理解Embeddings概念

在构建语义检索系统之前，我们需要理解什么是embeddings。简单来说，embeddings就是将文本转换成数值向量的过程，这些向量能够捕捉文本的语义信息。

语义相近的文本在向量空间中的距离会更近，这使得我们能够通过计算向量相似度来找到语义上相关的文档。这就是语义检索的核心原理。

3.2 使用Ollama生成Embeddings

Ollama提供了生成embeddings的接口，我们可以通过API调用来获取文本的向量表示：

import requests import json def generate_embedding(text, model="cogito:3b"): """使用Ollama生成文本嵌入向量""" url = "http://localhost:11434/api/embeddings" payload = { "model": model, "prompt": text } response = requests.post(url, json=payload) if response.status_code == 200: return response.json()["embedding"] else: raise Exception(f"生成嵌入失败: {response.text}") # 示例使用 text = "人工智能和机器学习的基本概念" embedding = generate_embedding(text) print(f"生成的向量维度: {len(embedding)}")

3.3 构建文档向量数据库

为了进行高效的语义检索，我们需要先构建一个文档向量数据库：

import numpy as np from sklearn.metrics.pairwise import cosine_similarity import pickle import os class VectorDatabase: def __init__(self): self.documents = [] self.embeddings = [] self.metadata = [] def add_document(self, text, metadata=None): """向数据库中添加文档""" embedding = generate_embedding(text) self.documents.append(text) self.embeddings.append(embedding) self.metadata.append(metadata or {}) def search(self, query, top_k=5): """语义搜索相似文档""" query_embedding = generate_embedding(query) query_embedding = np.array(query_embedding).reshape(1, -1) # 计算余弦相似度 similarities = cosine_similarity( query_embedding, np.array(self.embeddings) )[0] # 获取最相似的文档 indices = np.argsort(similarities)[::-1][:top_k] results = [] for idx in indices: results.append({ "document": self.documents[idx], "similarity": float(similarities[idx]), "metadata": self.metadata[idx] }) return results def save(self, filepath): """保存向量数据库""" with open(filepath, 'wb') as f: pickle.dump({ 'documents': self.documents, 'embeddings': self.embeddings, 'metadata': self.metadata }, f) def load(self, filepath): """加载向量数据库""" if os.path.exists(filepath): with open(filepath, 'rb') as f: data = pickle.load(f) self.documents = data['documents'] self.embeddings = data['embeddings'] self.metadata = data['metadata'] # 创建和使用向量数据库 db = VectorDatabase() db.add_document("人工智能是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的系统。") db.add_document("机器学习是人工智能的一个子集，它使计算机能够在没有明确编程的情况下从数据中学习。") db.add_document("深度学习是机器学习的一个分支，使用多层神经网络来处理复杂的数据模式。") # 保存数据库 db.save("knowledge_base.pkl")

4. 完整语义检索系统实现

4.1 构建端到端检索流程

现在我们将所有组件组合起来，构建一个完整的语义检索系统：

import time from typing import List, Dict class SemanticSearchSystem: def __init__(self, model_name="cogito:3b"): self.model_name = model_name self.vector_db = VectorDatabase() self.is_initialized = False def initialize_from_files(self, file_paths: List[str]): """从文本文件初始化知识库""" print("正在初始化知识库...") for file_path in file_paths: try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # 简单分段处理 paragraphs = [p for p in content.split('\n\n') if p.strip()] for i, para in enumerate(paragraphs): if len(para) > 50: # 只处理足够长的段落 self.vector_db.add_document( para, {"source": file_path, "paragraph": i} ) time.sleep(0.1) # 避免请求过快 except Exception as e: print(f"处理文件 {file_path} 时出错: {e}") self.is_initialized = True print(f"知识库初始化完成，共 {len(self.vector_db.documents)} 个文档") def search(self, query: str, top_k: int = 3) -> List[Dict]: """执行语义搜索""" if not self.is_initialized: raise Exception("请先初始化知识库") return self.vector_db.search(query, top_k) def interactive_search(self): """交互式搜索界面""" if not self.is_initialized: print("请先初始化知识库") return print("语义检索系统已启动，输入 'quit' 退出") while True: query = input("\n请输入搜索查询: ").strip() if query.lower() == 'quit': break if not query: continue start_time = time.time() results = self.search(query) search_time = time.time() - start_time print(f"\n找到 {len(results)} 个相关结果 (耗时: {search_time:.2f}秒):") for i, result in enumerate(results, 1): print(f"\n{i}. 相似度: {result['similarity']:.3f}") print(f"来源: {result['metadata'].get('source', '未知')}") print(f"内容: {result['document'][:200]}...") # 使用示例 if __name__ == "__main__": system = SemanticSearchSystem() # 假设我们有一些文本文件 text_files = ["ai_intro.txt", "ml_basics.txt", "deep_learning.txt"] # 初始化知识库 system.initialize_from_files(text_files) # 保存知识库以便后续使用 system.vector_db.save("ai_knowledge_base.pkl") # 执行搜索 results = system.search("什么是神经网络", top_k=3) for result in results: print(f"相似度: {result['similarity']:.3f}") print(f"内容: {result['document'][:100]}...") print("---")

4.2 高级检索功能增强

为了让语义检索系统更加实用，我们可以添加一些高级功能：

class EnhancedSemanticSearch(SemanticSearchSystem): def __init__(self, model_name="cogito:3b"): super().__init__(model_name) self.search_history = [] def advanced_search(self, query: str, top_k: int = 5, min_similarity: float = 0.3) -> List[Dict]: """增强版搜索，支持相似度阈值过滤""" results = self.search(query, top_k * 2) # 获取更多结果用于过滤 # 过滤低相似度结果 filtered_results = [ r for r in results if r['similarity'] >= min_similarity ][:top_k] # 记录搜索历史 self.search_history.append({ "query": query, "timestamp": time.time(), "results_count": len(filtered_results) }) return filtered_results def get_search_stats(self): """获取搜索统计信息""" total_searches = len(self.search_history) avg_results = (sum([h['results_count'] for h in self.search_history]) / total_searches) if total_searches > 0 else 0 return { "total_searches": total_searches, "average_results": avg_results, "recent_queries": [h['query'] for h in self.search_history[-5:]] } def batch_index_documents(self, documents: List[str], metadata_list: List[Dict] = None): """批量索引文档，提高效率""" if metadata_list is None: metadata_list = [{}] * len(documents) if len(documents) != len(metadata_list): raise ValueError("文档和元数据列表长度必须一致") print(f"开始批量索引 {len(documents)} 个文档...") for i, (doc, metadata) in enumerate(zip(documents, metadata_list)): if i % 10 == 0: print(f"已处理 {i}/{len(documents)} 个文档") self.vector_db.add_document(doc, metadata) time.sleep(0.05) # 稍微降低频率 self.is_initialized = True print("批量索引完成") # 使用增强版系统 enhanced_system = EnhancedSemanticSearch() # 批量添加文档 documents = [ "人工智能正在改变各行各业的工作方式。", "机器学习算法可以从数据中自动发现模式。", "深度学习使用神经网络处理复杂问题。", "自然语言处理让计算机能够理解人类语言。", "计算机视觉使机器能够看懂和理解图像。" ] enhanced_system.batch_index_documents(documents) # 执行高级搜索 results = enhanced_system.advanced_search( "机器学习应用", top_k=3, min_similarity=0.4 ) # 查看搜索统计 stats = enhanced_system.get_search_stats() print(f"总搜索次数: {stats['total_searches']}")

5. 实际应用案例与优化建议

5.1 构建专业领域知识库

语义检索系统特别适合构建专业领域的知识库。比如你可以创建一个技术文档检索系统：

def create_tech_knowledge_base(): """创建技术知识库示例""" system = EnhancedSemanticSearch() # 技术文档内容（实际应用中可以从文件读取） tech_documents = [ "Python是一种高级编程语言，以其简洁的语法和强大的库支持而闻名。", "Django是Python的一个高级Web框架，鼓励快速开发和简洁实用的设计。", "React是一个用于构建用户界面的JavaScript库，由Facebook开发和维护。", "Docker是一个开源平台，用于开发、交付和运行应用程序的容器化技术。", "Kubernetes是一个开源的容器编排系统，用于自动化应用程序部署、扩展和管理。", "TensorFlow是一个端到端的开源机器学习平台，由Google开发。", "PyTorch是一个开源的机器学习框架，以其动态计算图而受到研究人员欢迎。" ] metadata_list = [ {"category": "programming", "topic": "python"}, {"category": "web", "topic": "django"}, {"category": "web", "topic": "react"}, {"category": "devops", "topic": "docker"}, {"category": "devops", "topic": "kubernetes"}, {"category": "ai", "topic": "tensorflow"}, {"category": "ai", "topic": "pytorch"} ] system.batch_index_documents(tech_documents, metadata_list) system.vector_db.save("tech_knowledge_base.pkl") return system # 创建并使用技术知识库 tech_kb = create_tech_knowledge_base() results = tech_kb.advanced_search("Python编程语言", top_k=2) for result in results: print(f"相似度: {result['similarity']:.3f}") print(f"分类: {result['metadata']['category']}") print(f"内容: {result['document']}") print("---")

5.2 性能优化建议

在实际使用中，你可能需要考虑以下优化策略：

批量处理：一次性生成多个文档的embeddings，减少API调用次数
缓存机制：对频繁查询的结果进行缓存
向量索引：使用专业的向量数据库如FAISS或ChromaDB提高搜索效率
文档预处理：清理和标准化文本内容，提高embeddings质量

# 简单的缓存实现示例 from functools import lru_cache class CachedEmbeddingGenerator: def __init__(self, model_name="cogito:3b"): self.model_name = model_name @lru_cache(maxsize=1000) def get_cached_embedding(self, text: str): """带缓存的embeddings生成""" return generate_embedding(text, self.model_name) # 使用缓存 cached_generator = CachedEmbeddingGenerator() # 第一次调用会实际生成embedding embedding1 = cached_generator.get_cached_embedding("机器学习") # 第二次相同文本调用会直接从缓存获取 embedding2 = cached_generator.get_cached_embedding("机器学习") print("两次调用结果相同:", np.array_equal(embedding1, embedding2))

5.3 错误处理与稳定性增强

在实际生产环境中，需要添加完善的错误处理：

class RobustSemanticSearch(SemanticSearchSystem): def __init__(self, model_name="cogito:3b", max_retries=3): super().__init__(model_name) self.max_retries = max_retries def safe_generate_embedding(self, text: str): """带重试机制的embeddings生成""" for attempt in range(self.max_retries): try: return generate_embedding(text, self.model_name) except Exception as e: print(f"生成embedding失败 (尝试 {attempt + 1}/{self.max_retries}): {e}") if attempt == self.max_retries - 1: raise time.sleep(1) # 等待后重试 def safe_add_document(self, text: str, metadata=None): """安全的文档添加方法""" try: embedding = self.safe_generate_embedding(text) self.vector_db.documents.append(text) self.vector_db.embeddings.append(embedding) self.vector_db.metadata.append(metadata or {}) return True except Exception as e: print(f"添加文档失败: {e}") return False # 使用稳健版本 robust_system = RobustSemanticSearch() success = robust_system.safe_add_document("这是一个测试文档") if success: print("文档添加成功")