# 使用LangChain+LangGraph开发AI应用（11） - 案例：个人知识库


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017203759.png)

## 学习进度

这一篇做一个完整的项目，个人知识库。把你散落的 markdown 笔记吃进本地向量库，之后随时向它提问，它自己决定检索几次、检索什么，回答里的每个关键结论都带 `[文件名#块号]` 引用标记，顺着标记能翻回库里的原文块。

第 6 篇的 kb.py 是这个项目的雏形，检索几条固定段落拼进提示词，模型照单全收。这一篇把它升级成 Agent 版，代码在 `source/library.py`，三百行出头，命令行工具，三条子命令，ingest 导入文档，stats 看库存，ask 提问。演示笔记在 `source/notes/`，三个文件，仓库里直接能跑。

## 前言用途

问答类应用有两道坎。第一道是答非所问，模型对私域内容一无所知，靠提示词硬编只能救急。第二道是空口无凭，模型给了一个答案，你不知道它是真记得还是现编的，越是用它做决策，这个问题越致命。

RAG 解第一道，把资料检索进上下文。但第 6 篇那种固定 top-k 的做法有个结构性缺陷，检索几次、用什么词，全是代码写死的。问题问得笼统，一次检索的段落就不够；问题涉及两个主题，固定检索只能照顾到一个。引用溯源解第二道，答案里的每个结论都挂着出处，出处能翻回原文，编没编一眼便知。

这两个能力合起来，就是这一篇的全部。做完你手里有一个能日常用的笔记问答工具，更重要的是有一套可复用的结构，文档导入管道、检索工具、Agent 循环、引用回查，换到任何「资料问答」场景都是这几块积木。

## 原理概念

雏形和 Agent 版的差别在谁掌握检索的主动权。雏形是代码调一次 `similarity_search`，拿固定条数塞给模型，模型只负责组织语言。Agent 版把检索包成一个工具交给 create_agent，模型读你的问题，自己拟查询词，自己决定查一次还是查几次，看到检索结果不对路还会换个词再查。真实模型实测，问「怎么给模型挂工具，要注意什么」，它先查工具调用，扫了一眼结果又补了一轮错误处理的检索，两轮拿全材料才动笔。这个自主性就是 Agent 版的核心增量。

主动权交出去之后，新的问题是回答还信不信得过。解法是把「可追溯」做进数据结构里。每一块笔记入库时带一个确定性 id，规则就是 `文件名#块号`，检索工具把资料递给模型时，每段前面都挂上这个 id。系统提示词里要求模型引用资料时带上标记。回答回来，正则把标记抽出来，拿 id 回向量库查原文。三步全是确定性的，模型的自由发挥被圈在「引用哪个 id」这一件事上，引用错了 id，溯源清单立刻现形。

还有一个不必手写的东西，检索循环本身。自己写也不难，调一次检索、拼提示词、调模型，不满意再来一轮，第 6 篇的雏形就是这么干的。但换词重试、最多几轮、什么时候停，这些判断得自己写状态机，写完还要测。create_agent 把这套循环管了，模型每轮自己决定要不要再调工具，工具结果自动变 ToolMessage 回传，想限制轮数加一个参数，想看全程开 debug。案例里的检索工具只写了十行，循环、重试、停止条件全是现成的，这就是把检索交给 agent 的真正好处，业务代码里只剩业务。

导入管道的幂等性也压在 id 上。重复导入同一个文件，如果直接 add，块数翻倍，库里全是重复内容。给每块按 `文件名#块号` 生成 id 再 add，同 id 的块被覆盖，重导一遍等于原地刷新。但覆盖只救得了还存在的 id，文件改短之后旧的多余块没有新内容来覆盖，成了僵尸块。所以导入顺序是先删后加，按 metadata 里的 source 翻出这个文件现存的全部块 id，整批 delete，再写入新的。两次导入，块数永远等于当前文件实际该有的块数。

id 的形态是个刻意的选择。用文件名加块号，而不是内容 hash，权衡有三点。块号是人类可读的，`install.md#0` 一眼看出来自哪个文件的第几块，溯源清单打出来用户直接能对上；id 稳定，文件没改的块重导后 id 不变，向量库里省一次重算嵌入；同文件内容变了，块 id 依然成对出现，先删后加的替换逻辑写起来是整批操作，不用逐块比对 hash。内容 hash 的方案只在跨文档去重时有优势，个人笔记场景里这个需求几乎不出现，可读性优先。

![](https://static.xiongneng.me/library-arch-20261017203800.png)

## 核心用法

四块积木逐个过，代码都出自 `source/library.py`。

### 1. 文档导入管道

目录扫描、分割、确定性 id、替换式写入，一个函数收口。

```python
def ingest(paths: list[str]) -> None:
    store = load_store()
    for doc in raw_docs:
        source = doc.metadata["source"]
        old = chunk_ids_of(store, source)
        if old:
            store.delete(old)               # 先清这个文件的旧块
        chunks = SPLITTER.split_documents([doc])
        for i, c in enumerate(chunks):
            c.metadata["chunk"] = i
        store.add_documents(chunks,
                            ids=[f"{source}#{i}" for i in range(len(chunks))])
    store.dump(str(STORE_FILE))
```

```text
  install.md: 旧块 0 -> 新块 1
  tools.md: 旧块 0 -> 新块 1
  vector.md: 旧块 0 -> 新块 1
导入完成，替换 0 块，写入 3 块，库内现有 3 块
```

再跑一遍 ingest，输出变成替换 3 块、写入 3 块，库内还是 3 块，幂等得到实测。块 id 同时写进两个地方，InMemoryVectorStore 的主键位置和 metadata 的 source、chunk 字段，前者用来去重和回查，后者给 stats 和 list 做人读的展示。

管道的前半段是 load_docs 和分割器。目录扫描只认 txt 和 md 两种后缀，`rglob` 递归整棵目录树，排序保证多文件导入的顺序稳定。分割器沿用第 6 篇的配置，块 200 字、重叠 30 字，分隔符按中文排版习惯排了空行、换行、句号、逗号四级，标题和它的首段落在同一块的概率因此高了不少。chunk 字段是分割之后手工补进 metadata 的，split_documents 只继承原 metadata，块序号得自己数。

```python
def load_docs(path: Path) -> list[Document]:
    if path.is_file():
        files = [path] if path.suffix.lower() in SUFFIXES else []
    else:
        files = sorted(p for p in path.rglob("*") if p.suffix.lower() in SUFFIXES)
    return [Document(page_content=f.read_text(encoding="utf-8"),
                     metadata={"source": f.name}) for f in files]
```

文件名进 metadata 这一步越早越好，它在导入的那一刻就定死了后续一切溯源的锚点。

### 2. 检索工具

把相似度检索包成模型能调的工具，关键动作是给每段资料挂上来源标记。

```python
def format_hits(hits) -> str:
    if not hits:
        return "（没有检索到相关资料）"
    return "\n\n".join(
        f"[{d.metadata['source']}#{d.metadata['chunk']}] {d.page_content}"
        for d, _ in hits)

@tool
def search_notes(query: str, k: int = 3) -> str:
    """在个人笔记知识库里检索相关段落。

    返回带 [文件名#块号] 来源标记的资料，回答时引用这些标记。
    """
    return format_hits(store.similarity_search_with_score(query, k=k))
```

```text
[install.md#0] LangChain 用 pip 安装，命令是 pip install langchain...

[vector.md#0] 把文本切成小块，用嵌入模型算出向量存进向量库...
```

工具的 docstring 写给模型看，两句说清两件事，这个工具干什么、返回值里的标记怎么用。docstring 缺了后半句，模型大概率把标记当噪音丢掉，溯源就断了。检索结果为空也有兜底文案，模型看到它知道该换查询词而不是硬编。

### 3. Agent 问答

create_agent 把模型和检索工具接成循环，这一步的代码反而最短。

```python
def build_agent(store, model):
    return create_agent(model, tools=[search_notes], debug=False)

def ask(question: str) -> None:
    store = load_store()
    model = build_model() or fake_model_for(question, store)
    agent = build_agent(store, model)
    result = agent.invoke({"messages": [question]})
    answer = result["messages"][-1].content
```

result 的 messages 里能看到完整的协作过程，模型发起的每次 tool_calls、每条 ToolMessage 里的检索结果、最终回答。一次提问落下来的消息流长这样。

```text
[HumanMessage] 怎么给模型挂工具？要注意什么？
[AIMessage] （发起 tool_calls，query 为 怎么给模型挂工具）
[ToolMessage] [tools.md#0] # 工具调用备忘 给模型挂工具用 create_agent...
[AIMessage] （再发起一轮 tool_calls，query 为 工具调用注意事项）
[ToolMessage] （没有检索到相关资料）
[AIMessage] 给模型挂工具用 create_agent... [tools.md#0]
```

检索轮数的统计就是数 AIMessage 里带 tool_calls 的条数，这个数字直接打给用户看，自主检索不再是黑盒。第二条 ToolMessage 的空结果很有信息量，模型明确收到「没有检索到相关资料」，回答末尾那句「笔记里还没有相关细节」的诚实表态正是从这里来的。

本地没有 key 也能跑。fake_model_for 按问题先替模型检索一遍，编排出「调一次工具再作答」的两步脚本，行为和真实模型同构，测试全部压在这条路径上。

```python
def fake_model_for(question: str, store) -> ScriptedChatModel:
    probe = store.similarity_search_with_score(question, k=3)
    cites = [f"{d.metadata['source']}#{d.metadata['chunk']}" for d, _ in probe]
    body = "、".join(f"[{c}]" for c in cites[:2]) or "资料中没有直接相关的内容"
    return ScriptedChatModel(messages=iter([
        AIMessage(content="", tool_calls=[
            {"name": "search_notes", "args": {"query": question, "k": 3},
             "id": "t1", "type": "tool_call"}]),
        AIMessage(content=f"依据笔记，{question} 的要点整理如下，出处 {body}。"),
    ]))
```

替身引用的 id 来自真实检索的结果，溯源清单因此翻得回真原文，演示链路一步不缺。

### 4. 引用溯源

回答里的标记抽出来，回库翻原文，溯源清单就成了。

```python
CITE = re.compile(r"\[([^\[\]#]+#[0-9]+)\]")

def parse_citations(text: str) -> list[str]:
    seen, cites = set(), []
    for m in CITE.finditer(text):
        if m.group(1) not in seen:
            seen.add(m.group(1))
            cites.append(m.group(1))
    return cites

def source_trace(store, cites: list[str]) -> list[tuple[str, str]]:
    trace = []
    for cite in cites:
        item = store.store.get(cite)
        trace.append((cite, item["text"] if item else "（库内找不到该块）"))
    return trace
```

```text
回答: 依据笔记，LangChain 怎么安装？ 的要点整理如下，出处 [install.md#0]、[vector.md#0]。
（模型检索了 1 轮，回答带 2 处引用）
引用溯源:
  install.md#0 -> # LangChain 安装与初始化  LangChain 用 pip 安装，命令是 p
  vector.md#0 -> # 向量检索原理  把文本切成小块，用嵌入模型算出向量存进向量库
```

正则只认「非方括号内容加井号加数字」的形态，模型引用没写标记，或者写了 `[链接]` 这种普通方括号，都不会误伤。翻不到的 id 在清单里如实标注，不静默吞掉。--show-sources 开关把引用块的全文也展开，核对答案时不用再开笔记文件。

## 完整案例

四块积木拼起来的完整体验，三条命令。

```bash
uv run python library.py ingest notes/    # 导入，重复执行幂等
uv run python library.py list             # 库存清单
uv run python library.py ask "怎么给模型挂工具？要注意什么？"
```

```text
库内共 3 块
  1. [install.md#0] # LangChain 安装与初始化  LangChain 用 pip 安装，命令
  2. [tools.md#0] # 工具调用备忘  给模型挂工具用 create_agent，传 tools 列
  3. [vector.md#0] # 向量检索原理  把文本切成小块，用嵌入模型算出向量存进
```

list 是块级视角，stats 是文件级视角，导入管道有没有吃对东西，两个命令对着看一眼就够。接上真实模型（AGENT_MODEL 设为 `openai:deepseek-v4.1-flash`，走 YI 网关），问一个需要综合两处资料的问题。

```text
文件 3 个，共 3 块
  install.md: 1 块
  tools.md: 1 块
  vector.md: 1 块
```

接上真实模型（AGENT_MODEL 设为 `openai:deepseek-v4.1-flash`，走 YI 网关），问一个需要综合两处资料的问题。

```text
问题: 怎么给模型挂工具？要注意什么？
回答: 给模型挂工具用 create_agent，传 tools 列表 [tools.md#0]。要注意的几点：

1. docstring 要写清楚，模型靠它判断何时使用工具，还要说明参数含义；
2. 注意是否需要人工确认，涉及写操作的工具应当把确认环节放在工具外面。

---
笔记里还没有"工具报错怎么回传"的细节，如果你已经实践过，可以告诉我结论，我来补写。
（模型检索了 2 轮，回答带 1 处引用）
引用溯源:
  tools.md#0 -> # 工具调用备忘  给模型挂工具用 create_agent，传 tools 列表，模型
```

这段真实输出里有三个值得留意的细节。模型自己检索了两轮，先查怎么挂工具，看了一眼结果，又主动补了一轮检索去确认注意事项，主动权交接是真实发生的。回答末尾它指出笔记里缺少工具报错回传的内容，这是检索结果如实告诉它「没有相关资料」之后的诚实反应，比硬编一段强得多。溯源清单里 tools.md#0 翻出的原文正是回答第一句的依据，编造与引用的边界清清楚楚。

--show-sources 把核对做到极致，每个引用块的全文直接展开在溯源清单下面。

```text
引用溯源:
  tools.md#0 -> # 工具调用备忘  给模型挂工具用 create_agent，传 tools 列表，模型
      原文: # 工具调用备忘

      给模型挂工具用 create_agent，传 tools 列表，模型自己决定调不调、调哪个。

      工具函数写清 docstring 很关键，模型靠它判断什么时候该用这个工具。

      工具返回 ToolMessage 回传给模型，模型拿到结果再决定下一步。
```

答案对不对，扫一眼原文块就有结论，全程不用离开终端。

库存侧的两个命令各管一层，stats 聚合、list 明细。

```python
def stats() -> None:
    by_source: dict[str, int] = {}
    for item in store.store.values():
        src = item["metadata"].get("source", "?")
        by_source[src] = by_source.get(src, 0) + 1
```

实现就是遍历库内全部块，按 source 计数，二十行以内的函数，价值在养成导入后跑一下的习惯。块数骤降多半是源文件被误删或后缀没扫到，块数为零的知识库提问时只会得到一句「没有检索到相关资料」，提前在 stats 里看出来，省得去怀疑检索工具坏了。

测试在 `tests/test_library.py` 十一条，幂等导入、僵尸块清理、metadata 完整性、引用解析保序去重、溯源命中与翻空、替身问答、两轮检索计数、空库提示、按文件统计，全部走本地替身不花一分钱。加上前十篇的，全仓一百七十八条，全绿。

这个案例的扩展方向也留好了。换嵌入模型改 build_embeddings 一处，上持久化把 InMemoryVectorStore 换成任何 VectorStore 实现，接口不变；支持 pdf 和 docx 就往 load_docs 里加加载器；想把知识库挂进日常聊天，把 search_notes 注册进第 5 篇的任何 agent 就行，工具是自包含的。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangChain 官方文档 - RAG](https://docs.langchain.com/oss/python/langchain/rag)
- [LangChain 官方文档 - Agentic RAG](https://docs.langchain.com/oss/python/langchain/agentic-rag)
- [LangChain 官方文档 - Retrievers](https://docs.langchain.com/oss/python/langchain/retrievers)
- [LangGraph 官方文档 - Agents](https://docs.langchain.com/oss/python/langgraph/agents)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-11-knowledge-base/