MacBook上从零搞定LangChain：Python环境、ChromaDB持久化与OpenAI集成保姆级教程-编程实验室

MacBook上从零搞定LangChain：Python环境、ChromaDB持久化与OpenAI集成保姆级教程

1. 环境准备：打造MacOS专属Python开发环境

对于刚接触AI开发的Mac用户来说，环境配置往往是第一个拦路虎。不同于Windows系统，MacOS自带的Python版本通常较旧，且系统对Python环境的保护机制更为严格。我们先从最基础的Python环境搭建开始。

1.1 使用Homebrew安装Python3

Homebrew是MacOS上最受欢迎的包管理器，它能帮我们优雅地解决依赖问题。打开终端(Terminal)，执行以下命令安装Homebrew：

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

安装完成后，建议将Homebrew添加到PATH环境变量：

echo 'eval "$(/opt/homebrew/bin/brew shellenv)"' >> ~/.zshrc source ~/.zshrc

现在可以安装Python3了：

brew install python@3.10

注意：如果遇到Updating Homebrew...卡住的情况，可以按Control+C跳过更新直接安装。

安装完成后，验证Python版本：

python3 --version

1.2 配置Python环境变量

为了避免与系统Python冲突，我们需要明确指定使用Homebrew安装的Python。编辑shell配置文件：

nano ~/.zshrc

添加以下内容（路径可能因安装版本不同而有所变化）：

export PATH="/opt/homebrew/opt/python@3.10/bin:$PATH" alias python="/opt/homebrew/bin/python3" alias pip="/opt/homebrew/bin/pip3"

保存后执行：

source ~/.zshrc

1.3 解决常见依赖问题

在Mac上安装AI相关库时，可能会遇到HNSWLIB_NO_NATIVE=1错误。这是因为某些库需要编译原生扩展。提前设置这个环境变量可以避免问题：

echo 'export HNSWLIB_NO_NATIVE=1' >> ~/.zshrc source ~/.zshrc

2. 核心组件安装：LangChain与ChromaDB实战

2.1 安装LangChain生态

现在可以安装LangChain及其相关组件了。建议先升级pip：

python -m pip install --upgrade pip

然后安装核心库：

pip install langchain chromadb openai tiktoken

小技巧：如果下载速度慢，可以尝试使用清华镜像源：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple langchain chromadb

2.2 验证安装

创建一个简单的Python脚本test_env.py：

import langchain import chromadb print("LangChain版本:", langchain.__version__) print("ChromaDB版本:", chromadb.__version__)

运行脚本：

python test_env.py

如果能看到版本号输出，说明基础环境已就绪。

3. ChromaDB持久化存储实战

3.1 基础CRUD操作

让我们先了解ChromaDB的基本使用方法：

import chromadb # 创建内存中的客户端 client = chromadb.Client() # 创建集合(类似数据库表) collection = client.create_collection(name="test_collection") # 添加文档 collection.add( documents=["苹果是一种水果", "香蕉是热带水果", "汽车是交通工具"], metadatas=[{"type": "水果"}, {"type": "水果"}, {"type": "交通工具"}], ids=["id1", "id2", "id3"] ) # 查询 results = collection.query( query_texts=["水果"], n_results=2 ) print(results)

3.2 实现磁盘持久化

实际项目中，我们需要将向量数据库持久化到磁盘：

from chromadb.config import Settings # 持久化目录 persist_dir = "chroma_storage" # 配置持久化设置 client = chromadb.Client(Settings( persist_directory=persist_dir, chroma_db_impl="duckdb+parquet", )) # 创建持久化集合 collection = client.create_collection("persisted_data") # 添加数据 collection.add( documents=["Python是一种编程语言", "LangChain是AI框架", "ChromaDB是向量数据库"], ids=["doc1", "doc2", "doc3"] ) # 显式持久化到磁盘 client.persist() # 重新加载验证 new_client = chromadb.Client(Settings( persist_directory=persist_dir, chroma_db_impl="duckdb+parquet", )) loaded_collection = new_client.get_collection("persisted_data") print(loaded_collection.count()) # 应输出3

4. OpenAI集成与高级应用

4.1 配置OpenAI API

首先需要获取OpenAI API密钥，然后在代码中配置：

import os os.environ["OPENAI_API_KEY"] = "你的API密钥"

4.2 使用OpenAI Embeddings

LangChain提供了便捷的OpenAI集成方式：

from langchain.embeddings.openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings() # 生成文本的向量表示 text = "这是要向量化的文本" query_result = embeddings.embed_query(text) doc_result = embeddings.embed_documents([text]) print(f"查询向量长度: {len(query_result)}") print(f"文档向量长度: {len(doc_result[0])}")

4.3 完整知识库应用

结合LangChain和ChromaDB构建完整应用：

from langchain.document_loaders import TextLoader from langchain.text_splitter import CharacterTextSplitter from langchain.vectorstores import Chroma # 加载文档 loader = TextLoader("knowledge.txt") documents = loader.load() # 分割文档 text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200) texts = text_splitter.split_documents(documents) # 创建向量存储 vectorstore = Chroma.from_documents( documents=texts, embedding=OpenAIEmbeddings(), persist_directory="knowledge_db" ) # 持久化 vectorstore.persist() # 查询 query = "LangChain是什么？" docs = vectorstore.similarity_search(query) print(docs[0].page_content)

5. 性能优化与问题排查

5.1 提高ChromaDB性能

对于大型数据集，可以调整这些参数：

client = chromadb.Client(Settings( persist_directory="optimized_db", chroma_db_impl="duckdb+parquet", anonymized_telemetry=False, # 禁用遥测提升性能 allow_reset=True # 允许重置数据库 ))

5.2 常见错误解决

问题1：Illegal instruction: 4
解决方案：设置环境变量

export HNSWLIB_NO_NATIVE=1

问题2：ConnectionErrorwhen calling OpenAI API
解决方案：检查网络连接，或使用代理设置：

import openai openai.api_key = "你的API密钥" openai.proxy = "http://proxy.example.com:8080"

问题3：ValueError: Expected embedding dimension 1536 but got 768
解决方案：确保使用的嵌入模型一致：

OpenAIEmbeddings(model="text-embedding-ada-002")

6. 实际项目结构建议

一个规范的LangChain项目可以这样组织：

my_langchain_project/ ├── config/ │ └── settings.py # 存放API密钥等配置 ├── data/ │ ├── raw/ # 原始文档 │ └── processed/ # 处理后的数据 ├── db/ # 向量数据库存储 ├── src/ │ ├── loaders.py # 文档加载器 │ ├── embeddings.py # 嵌入处理 │ └── query.py # 查询逻辑 └── requirements.txt # 依赖列表

在settings.py中管理配置：

import os class Settings: OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "你的API密钥") PERSIST_DIR = os.path.join(os.path.dirname(__file__), "../db") settings = Settings()

7. 进阶技巧与最佳实践

7.1 批量处理大型文档

对于大型文档，可以使用以下优化策略：

from langchain.document_loaders import DirectoryLoader # 批量加载目录下的所有txt文件 loader = DirectoryLoader('./data/', glob="**/*.txt", loader_cls=TextLoader) # 使用多线程加载 documents = loader.load() # 更智能的文本分割 from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", "。", "？", "！", " ", ""] )

7.2 混合检索策略

结合多种检索方式提高准确性：

from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.vectorstores import Chroma # 创建向量检索器 vectorstore = Chroma(persist_directory="knowledge_db", embedding_function=OpenAIEmbeddings()) vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) # 创建BM25检索器 from langchain.retrievers import BM25Retriever bm25_retriever = BM25Retriever.from_documents(texts) bm25_retriever.k = 3 # 组合检索器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.5, 0.5] ) # 执行混合检索 results = ensemble_retriever.get_relevant_documents("查询问题")

7.3 监控与日志

添加日志记录有助于调试：

import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('langchain.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) try: # 你的LangChain代码 logger.info("操作成功完成") except Exception as e: logger.error(f"发生错误: {str(e)}", exc_info=True)