LangChain + Ollama：搭建员工手册 RAG 问答

给一份员工手册做问答：按 Markdown 小节切块，用 Ollama 的 qwen3-embedding 生成向量，存进 InMemoryVectorStore，再用 qwen2.5 1.5B 生成答案。 三个原则：

• 先评测检索（hit@k），再接生成；
• 答案必须带来源；模型忘了写，就由代码补上；
• 相似度太低就拒答。阈值根据实测的分数分布来定，并且承认它不是万能的。

先看清成品（box 上真实运行，手册内容是我编的“青禾科技”）

• 10 个问题：hit@1 和 hit@2 都是 100%。
• “在上海出差住酒店最多报多少钱一晚？”回答“最多报600元一晚。[来源: expense.md#差旅住宿]”。
• “公司食堂几点开饭？”的最高分是 0.370，低于阈值 0.5，直接拒答。
• “报销可以用支付宝吗？”的最高分是 0.52，越过了阈值，最后由模型按提示回答“手册里没有找到”。
• pytest 6 项通过，测试不需要 Ollama。

合规提醒：本系列只用本地 Ollama 模型，数据不出本机。如果换成云端模型，请先确认数据合规和服务商条款，API Key 放在环境变量里，不要共享账号。模型的许可证以它在 Ollama 或 Hugging Face 页面上写的为准。示例数据都是我自己编的。本文不列价格。

做完之后你会得到什么？

需要准备哪些环境？

安装：pip install -r requirements.txt

另外需要 numpy：InMemoryVectorStore 计算相似度要用到它，但 langchain-core 不会自动安装（见调试示例 1）。

文件结构

60-langchain-B-src/
├── docs/{leave,expense,it}.md
├── questions.json
├── rag.py
├── requirements.txt
└── test_rag.py

第 1 步：手册和问题集

# 报销制度（示例公司“青禾科技”，内容为教程虚构）

## 差旅住宿
一线城市住宿标准每晚不超过 600 元，其他城市不超过 400 元。超标部分自理，特殊情况需事先邮件申请。

## 交通
市内交通优先使用公共交通；晚上 22 点后加班可报销打车费，需在报销单中注明加班事由。

## 时限
费用发生后 60 天内提交报销，逾期不予受理。发票抬头必须为公司全称。

另外两份（leave.md、it.md）格式相同，见 src 目录。

[
 {"q": "入职三年有几天年假？", "src": "leave.md#年假"},
 {"q": "没休完的年假能留到明年吗？", "src": "leave.md#年假"},
 {"q": "病假请一周要谁批？", "src": "leave.md#病假"},
 {"q": "在上海出差住酒店最多报多少钱一晚？", "src": "expense.md#差旅住宿"},
 {"q": "加班晚了打车能报销吗？", "src": "expense.md#交通"},
 {"q": "发票过了三个月还能报吗？", "src": "expense.md#时限"},
 {"q": "密码多久要改一次？", "src": "it.md#密码"},
 {"q": "电脑丢了怎么办？", "src": "it.md#设备"},
 {"q": "想装个新软件需要什么流程？", "src": "it.md#软件"},
 {"q": "结婚能休几天假？", "src": "leave.md#婚假"}
]

第 2 步：RAG 代码

"""员工手册问答：LangChain + 本地 Ollama（嵌入 qwen3-embedding 0.6B，生成 qwen2.5 1.5B）。
- 按 Markdown 标题切块，每块带上 source = 文件名#小节标题，答案必须引用来源
- 先用问题集评测检索命中率（hit@k），再接生成
- 最高相似度低于阈值时拒答，不让模型凭空编"""
import json, os, pathlib, sys
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_text_splitters import MarkdownHeaderTextSplitter

EMBED = os.environ.get("EMBED_MODEL", "qwen3-embedding-local:0.6b")
LLM = os.environ.get("OLLAMA_MODEL", "qwen2.5-local:1.5b")
THRESHOLD = float(os.environ.get("RAG_THRESHOLD", "0.5"))

def load_chunks(folder="docs"):
    sp = MarkdownHeaderTextSplitter([("#", "title"), ("##", "section")], strip_headers=False)
    out = []
    for f in sorted(pathlib.Path(folder).glob("*.md")):
        for d in sp.split_text(f.read_text(encoding="utf-8")):
            if "section" in d.metadata:          # 只有大标题的那块（文件说明）不入库
                d.metadata["source"] = f"{f.name}#{d.metadata['section']}"
                out.append(d)
    return out

def build_store(chunks, embeddings=None):
    if embeddings is None:
        from langchain_ollama import OllamaEmbeddings
        embeddings = OllamaEmbeddings(model=EMBED)
    return InMemoryVectorStore.from_documents(chunks, embeddings)

def hit_at_k(store, questions, k=2):
    hits, misses = 0, []
    for item in questions:
        got = [d.metadata["source"] for d in store.similarity_search(item["q"], k=k)]
        if item["src"] in got:
            hits += 1
        else:
            misses.append((item["q"], item["src"], got))
    return hits / len(questions), misses

PROMPT = ChatPromptTemplate.from_messages([
    ("system", "只根据下面的资料回答，用中文，一两句话。回答末尾用 [来源: xxx] 标出依据。资料里没有答案就说“手册里没有找到”。\n\n{context}"),
    ("human", "{q}"),
])

def answer(store, q, llm=None, k=2):
    scored = store.similarity_search_with_score(q, k=k)
    if not scored or scored[0][1] < THRESHOLD:
        return {"answer": "手册里没有找到相关内容。", "sources": [], "top_score": scored[0][1] if scored else None}
    if llm is None:
        from langchain_ollama import ChatOllama
        llm = ChatOllama(model=LLM, temperature=0)
    ctx = "\n\n".join(f"[来源: {d.metadata['source']}]\n{d.page_content}" for d, _ in scored)
    msg = (PROMPT | llm).invoke({"context": ctx, "q": q})
    text = msg.content.strip()
    if "[来源" not in text and "没有找到" not in text:      # 小模型常忽略引用要求：由代码补上，不依赖模型
        text += f" [来源: {scored[0][0].metadata['source']}]"
    return {"answer": text, "sources": [d.metadata["source"] for d, _ in scored], "top_score": scored[0][1]}

if __name__ == "__main__":
    chunks = load_chunks(); store = build_store(chunks)
    qs = json.load(open("questions.json", encoding="utf-8"))
    for k in (1, 2):
        rate, misses = hit_at_k(store, qs, k)
        print(f"hit@{k} = {rate:.0%}", *(f"\n  未命中：{q} 期望 {s} 实得 {g}" for q, s, g in misses))
    for q in sys.argv[1:] or ["在上海出差住酒店最多报多少钱一晚？", "公司食堂几点开饭？", "报销可以用支付宝吗？"]:
        r = answer(store, q)
        print(f"\nQ: {q}\nA: {r['answer']}\n   top_score={r['top_score']:.3f} sources={r['sources']}")

第 3 步：测试

import json
from langchain_core.embeddings import Embeddings
from langchain_core.language_models.fake_chat_models import FakeListChatModel
import rag

class CharEmbeddings(Embeddings):
    """测试用：按若干关键字的出现次数做向量，确定性、无需 Ollama。"""
    KEYS = ["年假", "病假", "婚", "住宿", "酒店", "打车", "报销", "密码", "电脑", "设备", "软件", "发票"]
    def _v(self, t):
        v = [float(t.count(k)) for k in self.KEYS]
        return v + [0.0 if any(v) else 1.0]          # 一个关键字都不含：放到单独的“其他”维度
    def embed_documents(self, texts):
        return [self._v(t) for t in texts]
    def embed_query(self, text):
        return self._v(text)

def store():
    return rag.build_store(rag.load_chunks(), CharEmbeddings())

def test_chunks_have_sources():
    srcs = [d.metadata["source"] for d in rag.load_chunks()]
    assert len(srcs) == 9 and "expense.md#差旅住宿" in srcs and not any(s.endswith("#") for s in srcs)

def test_hit_at_k_reports_misses():
    rate, misses = rag.hit_at_k(store(), [{"q": "年假", "src": "leave.md#年假"}, {"q": "食堂", "src": "x.md#y"}], k=1)
    assert rate == 0.5 and misses[0][0] == "食堂"

def test_refuse_below_threshold():
    r = rag.answer(store(), "食堂几点开饭", llm=FakeListChatModel(responses=["不该被调用"]))
    assert r["sources"] == [] and "没有找到" in r["answer"]

def test_citation_added_when_model_omits_it():
    r = rag.answer(store(), "酒店住宿标准", llm=FakeListChatModel(responses=["每晚不超过 600 元。"]))
    assert r["answer"].endswith("[来源: expense.md#差旅住宿]")

def test_model_citation_kept():
    r = rag.answer(store(), "酒店住宿标准", llm=FakeListChatModel(responses=["600 元 [来源: expense.md#差旅住宿]"]))
    assert r["answer"].count("[来源") == 1

def test_questions_file_sources_exist():
    srcs = {d.metadata["source"] for d in rag.load_chunks()}
    assert all(q["src"] in srcs for q in json.load(open("questions.json", encoding="utf-8")))

第 4 步：运行

$ python rag.py
hit@1 = 100%
hit@2 = 100%

Q: 在上海出差住酒店最多报多少钱一晚？
A: 在上海出差住酒店最多报600元一晚。 [来源: expense.md#差旅住宿]
   top_score=0.648 sources=['expense.md#差旅住宿', 'expense.md#交通']

Q: 公司食堂几点开饭？
A: 手册里没有找到相关内容。
   top_score=0.370 sources=[]

Q: 报销可以用支付宝吗？
A: 手册里没有找到。
   top_score=0.521 sources=['expense.md#时限', 'expense.md#交通']
$ python -m pytest -q
6 passed

同一个问题在不同次运行中的分数会有千分位的差别（实测 0.646 和 0.648），设阈值时要留出余量。

调试示例

1. 缺少 numpy。 第一次运行时，在检索这一步报错 ImportError: cosine_similarity requires numpy to be installed。langchain-core 把 numpy 当作可选依赖，所以要自己安装（box 上装的是 2.5.4）。我已经把它写进了 requirements.txt。

2. 阈值是怎么定的，又挡不住什么。 我实测了 10 个手册内问题的最高分，范围是 0.537 到 0.743；5 个手册外问题里，食堂、年会、天气、停车位的分数在 0.289 到 0.374 之间，但“报销可以用支付宝吗？”有 0.516，因为它和“报销”小节字面上很接近。所以 0.5 这个阈值只能挡住明显无关的问题，剩下的要靠提示词里的“资料里没有就说没找到”。实测 qwen2.5 1.5B 对这个问题确实回答了“手册里没有找到”。

3. 小模型不写引用。 提示词要求在末尾标出 [来源: xxx]，但 qwen2.5 1.5B 的回答是“最多报600元一晚。”，没有写来源。现在改为由代码补上最相关小节的来源；模型自己写了来源时不重复添加（test_model_citation_kept）。

4. 测试替身自己有 bug。 关键字向量的第一版，对不含任何关键字的问题返回 [0, …, 0, 1e-6]，结果这个微小的分量正好落在“发票”这一维，和“时限”小节的相似度变成 1.0，test_refuse_below_threshold 因此失败。我给这类问题单独加了一个“其他”维度，测试随之通过。

常见问题

模板规则

• 每个块都带 source。
• 先看 hit@k 和分数分布，再定阈值。
• 引用由代码保证。
• 换嵌入模型后重新评测，并重新确定阈值。

项目资源

RAG 先把检索做对、做可测，生成只是最后一步。