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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

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))

1
2
3
4

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

1
2
3
4
5
6
7

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]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

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))

1
2
3

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

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

1
2
3
4
5
6
7
8
9

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

1
2
3
4
5
6

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

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}。"),
    ]))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

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

1
2
3
4
5

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

1
2
3

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

1
2
3
4

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

1
2
3
4

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

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

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

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

1
2
3
4
5
6
7
8
9

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

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

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

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

1
2
3
4
5

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