有了我们的Docker部署做先行军,后面我们就可以分块了解RAG项目中各个必不可少的tools,本篇主要讲的是ollama➕logging➕bm25!!!

前言

在构建RAG（检索增强生成）系统的过程中，我逐渐认识到一个事实：RAG系统就是被各种零散的工具组合起来的。今天，我们聚焦三个核心工具：ollama（本地大模型部署）、logging（日志监控）和BM25（传统检索算法）。

这篇文章记录了我的学习过程，包含代码实践、难点分析和深度讨论。

---

一、Ollama：本地大模型的部署与调用

1.1 为什么选择Ollama？

在RAG系统中，大模型负责最终的答案生成。Ollama让我们能够在本地运行开源模型，具有以下优势：

• 数据安全：不需要将敏感数据发送到云端

• 成本可控：无需按token付费

• 离线可用：不依赖网络连接

1.2 四种调用模式详解

import ollama
from ollama import Client
import requests

模式1：基础聊天模式

def chatMode():
    response = ollama.chat(
        model='qwen2.5:0.5b',
        messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}]
    )
    print(response['message']['content'])

重难点分析：

• model参数：需要提前通过ollama pull qwen2.5:0.5b下载模型

• messages格式：遵循OpenAI的对话格式，支持system/user/assistant三种角色

• 返回值结构：response['message']['content']才是模型输出内容

模式2：流式输出

def streamMode():
    stream = ollama.chat(
        model='qwen2.5:0.5b',
        messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}],
        stream=True,
    )
    for chunk in stream:
        print(chunk['message']['content'], end='', flush=True)

重难点分析：

• flush=True：强制刷新缓冲区，实现print实时输出

• 流式输出的本质：逐token返回，而不是等待完整响应,交互更加友好

• 用户体验：大模型响应慢时，流式输出能让用户看到"正在思考"的过程

思考：

问：流式输出和普通输出有什么区别？

答：普通模式是"等菜全部做好再端上来"，流式模式是"边做边端，做好一道上一道"。对于大模型这种响应时间较长的场景，流式输出能显著提升用户体验。

模式3：远程调用

def remoteMode():
    client = Client(host='http://127.0.0.1:11434')
    response = client.chat(
        model='qwen2.5:0.5b',
        messages=[{'role': 'user', 'content': '为什么天空是蓝色的？'}]
    )
    print(response['message']['content'])

重难点分析：

• 默认端口：Ollama服务默认运行在11434端口

• 远程调用场景：模型在GPU服务器上，应用在其他机器上

• 与本地调用的区别：只是多了网络传输开销

模式4：原始HTTP请求

def rawMode():
    url = "http://127.0.0.1:11434/api/chat"
    data = {
        "model": "qwen2.5:0.5b",
        "options": {"temperature": 0.5},
        "stream": False,
        "messages": [{"role": "system", "content": "你是谁？"}]
    }
    response = requests.post(url, json=data, timeout=60)
    print(response.json()['message']['content'])

重难点分析：

• temperature参数：控制输出的随机性，0-1之间，越低越确定

• timeout=60：必须设置！否则网络问题会导致程序卡死

• 这种方式的优势：可以自定义更多参数（top_p、max_tokens等）

1.3 实践心得

模式	适用场景	优点	缺点
chatMode	简单问答	代码简洁	不支持流式
streamMode	需要实时反馈	用户体验好	处理稍复杂
remoteMode	分布式部署	灵活	有网络延迟
rawMode	高度定制	参数完整	代码冗长

---

二、Logging：RAG系统的可观测性

2.1 为什么需要日志？

在RAG系统中，一个请求可能经过：检索 → 重排 → 生成等多个环节。没有日志，你就像在黑暗中摸索。

2.2 四种日志配置详解

基础配置

def basicLog():
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger("Example1")
    
    logger.debug("调试信息")      # 不会输出（级别不够）
    logger.info("程序运行正常")   # ✅ 会输出
    logger.warning("注意问题")    # ✅ 会输出
    logger.error("发生错误")      # ✅ 会输出

重难点分析：

• 日志级别：DEBUG < INFO < WARNING < ERROR < CRITICAL

• basicConfig只能配置一次，后续调用basicConfig重新配置无效

  basicConfig() 是一次性初始化工具，不是动态配置工具。可以考虑为根logger

  命名 logger 有自己的配置用自己的配置,没有的话继承自根logger

  根logger只在第一次配置的适合生效

  核心就是：继承 + 可覆盖 + 一次性初始化。

  如果你需要动态调整日志级别、格式等，应该直接操作 logger 和 handler，而不是依赖多次调用 basicConfig。

带格式的日志

def withFormat():
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger("Example2")
    logger.debug("调试模式已开启")

格式化占位符说明：

占位符	含义	示例
%(asctime)s	时间	2024-01-15 10:30:45,123
%(levelname)s	日志级别	INFO
%(message)s	日志内容	程序运行正常
%(name)s	记录器名称	Example2
%(filename)s	文件名	app.py

输出到文件

def save2File():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        filename='app.log',
        filemode='a',  # 'a':追加, 'w':覆盖
        encoding='utf-8'
    )

重难点分析：

• filemode='a'：追加模式，程序重启不会覆盖之前的日志

• 重要：一旦指定了filename，控制台就不再显示日志

• encoding='utf-8'：处理中文日志必须设置

同时输出到文件和控制台

def consoleAndFile():
    logger = logging.getLogger("Example4")
    logger.setLevel(logging.DEBUG)
    
    # 控制台处理器：只显示INFO及以上
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    
    # 文件处理器：记录所有DEBUG及以上
    file_handler = logging.FileHandler('app.log', mode='a', encoding='utf-8')
    file_handler.setLevel(logging.DEBUG)
    
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    console_handler.setFormatter(formatter)
    file_handler.setFormatter(formatter)
    
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)

思考：

问：如果配置了写入文件，不单独配置显示控制台的话，控制台就不显示了是吧？

答：完全正确！Python logging的设计哲学是：当你明确要把日志写入文件时，说明你不需要在控制台看日志了（比如生产环境）；而在开发调试时，通常只看控制台就够了。

这个讨论让我理解了logging的设计思想：显式优于隐式。

2.3 RAG系统中的日志实践

class RAGSystem:
    def __init__(self):
        self.logger = self._setup_logger()
    
    def _setup_logger(self):
        logger = logging.getLogger("RAG")    #创建/获取一个命名 logger 实体
        logger.setLevel(logging.DEBUG)
        
        # 文件记录所有
        fh = logging.FileHandler('rag.log')
        fh.setLevel(logging.DEBUG)
        
        # 控制台只记录重要信息
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        
        logger.addHandler(fh)
        logger.addHandler(ch)
        return logger
    
    def retrieve(self, query):
        self.logger.info(f"检索查询: {query}")
        # 检索逻辑
        self.logger.debug(f"检索结果: {results}")
        return results

---

三、BM25：传统检索算法

"""
TF-IDF = TF × IDF

TF = 词在文档中出现次数 / 文档总词数
IDF = log(总文档数 / 包含该词的文档数)

BM25 = IDF × (TF × (k1 + 1)) / (TF + k1 × (1 - b + b × 文档长度 / 平均长度))

IDF = log((总文档数 - 包含该词的文档数 + 0.5) / (包含该词的文档数 + 0.5))

"""

TF-IDF

• ✅ 简单、计算快、易理解

• ❌ 词频线性增长不合理（100次≠100倍相关）

• ❌ 不考虑文档长度

BM25

• ✅ 词频饱和（出现10次后增益递减）

• ✅ 长度归一化（短文档优先）

• ✅ 可调参数适配不同场景

• ❌ 计算稍复杂

BM25 = TF-IDF的进化版，解决了词频过饱和和文档长度偏差问题，是搜索引擎的默认排序算法。

3.1 为什么还要学BM25？

在向量检索盛行的今天，BM25仍然有它的价值：

• 可解释性强：关键词匹配，容易理解为什么返回这个结果

• 计算资源少：不需要GPU，纯CPU计算

• 作为Baseline：评估向量检索效果的基准

3.2 BM25核心代码解析

import jieba
from rank_bm25 import BM25L

class BM25Search():
    def __init__(self, documents):
        self.documents = documents
        # 中文分词：这是关键步骤
        self.tokenized_docs = [jieba.lcut(doc) for doc in documents]
        self.bm25 = BM25L(self.tokenized_docs)
    
    def search(self, query):
        tokenized_query = jieba.lcut(query)
        try:
            scores = self.bm25.get_scores(tokenized_query)
            best_idx = scores.argmax()
            best_score = scores[best_idx]
            best_doc = self.documents[best_idx]
            return best_doc, best_score
        except Exception as e:
            return None, 0

3.3 重难点深度解析

为什么用jieba.lcut？

BM25是基于词袋模型的算法，需要先对中文进行分词。jieba.lcut返回的是分词列表：

import jieba
text = "我喜欢编程"
print(jieba.lcut(text))  # ['我', '喜欢', '编程']

为什么选择BM25L而不是BM25？

变体	特点	适用场景
BM25	基础版本	标准场景
BM25L	处理长文档更好	文档长度差异大时
BM25Plus	处理短查询更好	查询词少时

异常处理的重要性

try:
    scores = self.bm25.get_scores(tokenized_query)
except Exception as e:
    return None, 0  # 返回None而不是崩溃

3.4 测试示例

documents = ["我喜欢编程", "编程很有趣", "我吃了一个苹果"]
bm25_model = BM25Search(documents)
query = "他从前喜欢编程"
result, score = bm25_model.search(query)
print(result)  # "我喜欢编程"
print(score)   # 0.98（相似度分数）

3.5 BM25 vs 向量检索的讨论

问：在RAG系统中，BM25和向量检索应该怎么选择？

答：不是选择问题，是结合问题。混合检索（Hybrid Search）同时使用BM25和向量检索，然后融合排序。BM25擅长精确关键词匹配，向量检索擅长语义匹配，两者互补。

这让我理解了为什么现代RAG系统通常采用混合检索策略。

---

四、三个工具的协同：RAG系统雏形

class SimpleRAG:
    def __init__(self):
        self.logger = self._setup_logger()
        self.bm25 = BM25Search(documents)
        self.llm_client = ollama.Client()
    
    def query(self, question):
        # 1. 日志记录
        self.logger.info(f"用户问题: {question}")
        
        # 2. BM25检索
        doc, score = self.bm25.search(question)
        self.logger.debug(f"检索得分: {score}, 文档: {doc}")
        
        # 3. 构建Prompt
        prompt = f"基于以下内容回答问题：\n{doc}\n\n问题：{question}"
        
        # 4. 调用大模型生成
        response = self.llm_client.chat(
            model='qwen2.5:0.5b',
            messages=[{'role': 'user', 'content': prompt}]
        )
        
        self.logger.info("生成完成")
        return response['message']['content']

---

写在最后：RAG系统的学习就像上一趟快车，工具层出不穷。但掌握了核心工具和思想，就能以不变应万变。希望这篇文章能帮助你在RAG的学习道路上少走一些弯路。

如果你在实践过程中遇到问题，欢迎交流讨论！