{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 24_项目实战一:个人知识库问答助手\n", "\n", "## 学习目标\n", "1. 综合运用 LangChain 和 LangGraph 知识解决实际问题\n", "2. 掌握项目需求分析、模块划分和技术选型\n", "3. 开始构建第一个综合项目:个人知识库问答助手\n", "\n", "本节课会从项目角度出发,带大家一步步设计一个“个人知识库问答助手”。它的核心能力是:**用户上传或准备一些资料,助手根据资料内容回答问题,而不是凭空编答案**。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. 项目背景:为什么需要个人知识库问答助手\n", "\n", "日常学习和工作中,我们经常会有很多资料,例如:\n", "\n", "- 课程笔记\n", "- 项目文档\n", "- 公司制度\n", "- 技术文章\n", "- 会议记录\n", "\n", "如果资料很多,直接人工查找会很慢。个人知识库问答助手要解决的问题就是:\n", "\n", "> 用户提出问题,系统先从资料中找到相关内容,再基于这些内容生成回答。\n", "\n", "这类系统通常使用 RAG 技术。RAG 的全称是 Retrieval-Augmented Generation,中文可以理解为“检索增强生成”。\n", "\n", "通俗地说:\n", "\n", "1. 先查资料\n", "2. 再带着资料回答\n", "\n", "这样可以减少大模型胡编乱造的问题。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. 需求分析\n", "\n", "做项目之前,不要急着写代码。第一步应该先想清楚:这个项目到底要解决什么问题。\n", "\n", "本项目的基础需求如下:\n", "\n", "| 需求 | 说明 |\n", "| --- | --- |\n", "| 导入资料 | 准备一批文本资料,作为知识库内容 |\n", "| 文本切分 | 把长文档切成小段,方便检索 |\n", "| 内容检索 | 根据用户问题找到最相关的资料片段 |\n", "| 生成回答 | 根据检索到的资料组织答案 |\n", "| 返回依据 | 告诉用户答案主要参考了哪些内容 |\n", "\n", "第一版项目先做最小可用版本,也就是 MVP。我们先不处理复杂文件上传,也不接入真实数据库,而是用少量文本模拟知识库。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. 项目整体流程\n", "\n", "个人知识库问答助手的流程可以画成下面这样:\n", "\n", "```text\n", "准备文档 -> 切分文档 -> 建立索引 -> 用户提问 -> 检索相关片段 -> 生成答案\n", "```\n", "\n", "如果用生活中的例子理解:\n", "\n", "- 准备文档:把书放进书架\n", "- 切分文档:给书按章节拆开\n", "- 建立索引:做目录和标签\n", "- 用户提问:用户问“某个知识点是什么”\n", "- 检索相关片段:先翻目录找到相关章节\n", "- 生成答案:根据章节内容整理成自然语言回答\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. 技术选型\n", "\n", "在真实项目中,可以使用下面这些技术:\n", "\n", "| 模块 | 可选技术 | 作用 |\n", "| --- | --- | --- |\n", "| 文档加载 | LangChain DocumentLoader | 读取 PDF、Markdown、网页等资料 |\n", "| 文本切分 | RecursiveCharacterTextSplitter | 把长文本切成小块 |\n", "| 向量模型 | Embeddings | 把文本转成向量 |\n", "| 向量数据库 | FAISS、Chroma | 存储并检索相似文本 |\n", "| 大模型 | ChatOpenAI 或其他模型 | 根据资料生成答案 |\n", "| 流程编排 | LangGraph | 把检索、回答、检查等步骤组织成工作流 |\n", "\n", "为了让代码更容易运行,本节先用纯 Python 实现一个简化版 RAG。理解原理后,再过渡到 LangChain 和 LangGraph。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. 第一步:准备模拟知识库\n", "\n", "我们先不用读取外部文件,而是在代码中准备几段文本,模拟个人知识库中的资料。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "documents = [\n", " \"LangChain 是一个用于开发大模型应用的框架,它提供模型调用、提示词模板、文档加载、检索问答等能力。\",\n", " \"LangGraph 是 LangChain 生态中的流程编排框架,适合构建多步骤、有状态、可分支的智能体应用。\",\n", " \"RAG 是检索增强生成技术,核心流程是先从知识库中检索相关资料,再让大模型基于资料生成答案。\",\n", " \"个人知识库问答助手可以帮助用户从自己的笔记、文档和资料中快速找到答案,减少人工查找成本。\",\n", " \"向量数据库用于存储文本向量,并根据相似度快速找到与用户问题最相关的文本片段。\",\n", "]\n", "\n", "print(f\"知识库中共有 {len(documents)} 条资料\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码完成了知识库的准备工作:\n", "\n", "1. `documents` 是一个列表,列表里的每个字符串代表一条资料。\n", "2. 这里的资料围绕 LangChain、LangGraph、RAG、个人知识库和向量数据库。\n", "3. `len(documents)` 用来统计资料数量。\n", "4. 在真实项目中,`documents` 通常不是手写的,而是从 PDF、Markdown、Word、网页或数据库中读取出来的。\n", "\n", "这一步对应项目流程中的“准备文档”。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. 第二步:文本切分\n", "\n", "真实文档可能很长,如果直接把整篇文档拿去检索,效果通常不好。\n", "\n", "所以我们需要把长文本切成较小的片段。这里用一个简单函数演示文本切分。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def split_text(text, chunk_size=30):\n", " chunks = []\n", " for start in range(0, len(text), chunk_size):\n", " chunk = text[start:start + chunk_size]\n", " chunks.append(chunk)\n", " return chunks\n", "\n", "\n", "all_chunks = []\n", "for doc in documents:\n", " chunks = split_text(doc, chunk_size=30)\n", " all_chunks.extend(chunks)\n", "\n", "for index, chunk in enumerate(all_chunks, start=1):\n", " print(f\"片段 {index}: {chunk}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码实现了一个简化版文本切分器:\n", "\n", "1. `split_text(text, chunk_size=30)` 表示把文本按固定长度切分,每段最多 30 个字符。\n", "2. `range(0, len(text), chunk_size)` 会生成每个片段的起始位置。\n", "3. `text[start:start + chunk_size]` 使用字符串切片取出一小段文本。\n", "4. `all_chunks.extend(chunks)` 把每篇文档切出来的片段合并到总列表中。\n", "5. 最后的循环会打印所有文本片段,方便观察切分效果。\n", "\n", "真实项目中,通常会使用 LangChain 的 `RecursiveCharacterTextSplitter`,它比这里的固定长度切分更智能,会尽量按段落、句子等边界切分。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. 第三步:实现一个简单检索器\n", "\n", "标准 RAG 通常会用向量相似度检索。为了让初学者更容易理解,我们先用“关键词重合数量”模拟检索。\n", "\n", "规则很简单:用户问题和某个资料片段中重复出现的字越多,就认为它越相关。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def simple_score(question, chunk):\n", " question_chars = set(question)\n", " chunk_chars = set(chunk)\n", " common_chars = question_chars & chunk_chars\n", " return len(common_chars)\n", "\n", "\n", "def retrieve(question, chunks, top_k=2):\n", " scored_chunks = []\n", " for chunk in chunks:\n", " score = simple_score(question, chunk)\n", " scored_chunks.append((score, chunk))\n", "\n", " scored_chunks.sort(reverse=True, key=lambda item: item[0])\n", " return scored_chunks[:top_k]\n", "\n", "\n", "question = \"什么是 RAG?\"\n", "retrieved_chunks = retrieve(question, all_chunks, top_k=3)\n", "\n", "for score, chunk in retrieved_chunks:\n", " print(f\"相关度分数:{score},内容:{chunk}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码实现了一个简单检索器:\n", "\n", "1. `simple_score(question, chunk)` 用来计算问题和资料片段的相关度。\n", "2. `set(question)` 会把问题中的字符去重后放入集合。\n", "3. `question_chars & chunk_chars` 表示取两个集合的交集,也就是问题和片段中共同出现的字符。\n", "4. `len(common_chars)` 用共同字符数量作为相关度分数。\n", "5. `retrieve(question, chunks, top_k=2)` 会给所有片段打分,并返回分数最高的前 `top_k` 个。\n", "6. `scored_chunks.sort(reverse=True, key=lambda item: item[0])` 表示按照分数从高到低排序。\n", "\n", "这个检索器很简单,但它帮助我们理解了检索阶段的本质:**从大量资料中找出最可能有用的内容**。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 8. 第四步:根据检索结果生成答案\n", "\n", "在真实 RAG 系统中,生成答案通常由大模型完成。\n", "\n", "这里为了避免依赖外部 API,我们先用一个函数模拟回答生成:把检索到的资料整理成答案。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def generate_answer(question, retrieved_chunks):\n", " context = \"\".join([chunk for score, chunk in retrieved_chunks])\n", " answer = f\"问题:{question}\\n\\n\"\n", " answer += \"根据知识库中检索到的内容,可以参考以下信息回答:\\n\"\n", " answer += context\n", " answer += \"\\n\\n总结:以上内容是从个人知识库中找到的相关资料,回答时应优先依据这些资料。\"\n", " return answer\n", "\n", "\n", "answer = generate_answer(question, retrieved_chunks)\n", "print(answer)\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码模拟了回答生成过程:\n", "\n", "1. `generate_answer(question, retrieved_chunks)` 接收用户问题和检索结果。\n", "2. `[chunk for score, chunk in retrieved_chunks]` 从检索结果中取出文本片段,忽略分数。\n", "3. `'\\n'.join(...)` 把多个片段用换行符拼接成上下文。\n", "4. `answer += ...` 逐步拼接最终回答内容。\n", "5. 真实项目中,这一步通常会把 `question` 和 `context` 放进提示词,然后交给大模型生成更自然、更完整的回答。\n", "\n", "这一阶段的关键原则是:**回答要基于检索到的资料,而不是凭空生成**。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 9. 封装成一个完整问答函数\n", "\n", "现在我们已经有了文档片段、检索器和回答生成函数。下一步把它们封装成一个完整的 `ask_knowledge_base` 函数。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def ask_knowledge_base(question, chunks, top_k=3):\n", " retrieved = retrieve(question, chunks, top_k=top_k)\n", " answer = generate_answer(question, retrieved)\n", " return {\n", " \"question\": question,\n", " \"retrieved\": retrieved,\n", " \"answer\": answer,\n", " }\n", "\n", "\n", "result = ask_knowledge_base(\"LangGraph 适合做什么?\", all_chunks)\n", "print(result[\"answer\"])\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码把前面的步骤组合成一个完整功能:\n", "\n", "1. `ask_knowledge_base` 是问答助手的主入口。用户只需要传入问题。\n", "2. `retrieve(question, chunks, top_k=top_k)` 先从知识库片段中找相关内容。\n", "3. `generate_answer(question, retrieved)` 再根据检索结果生成答案。\n", "4. 函数返回一个字典,里面包含原问题、检索结果和最终答案。\n", "5. 返回检索结果的好处是:后续可以展示“参考来源”,让用户知道答案从哪里来。\n", "\n", "到这里,我们已经完成了一个最小版本的个人知识库问答助手。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 10. 用 LangChain 思路改造项目结构\n", "\n", "虽然上面的代码是纯 Python,但它已经对应了 LangChain RAG 的主要模块。\n", "\n", "| 当前示例 | LangChain 中常见模块 |\n", "| --- | --- |\n", "| `documents` | DocumentLoader 加载出的文档 |\n", "| `split_text` | TextSplitter |\n", "| `retrieve` | Retriever |\n", "| `generate_answer` | LLMChain 或 LCEL 链 |\n", "| `ask_knowledge_base` | 完整 RAG Chain |\n", "\n", "所以学习项目时不要只记 API,更重要的是理解每个模块承担什么职责。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 11. LangChain 版伪代码\n", "\n", "下面给出一个接近真实项目的 LangChain 写法。由于不同环境中的模型和 API Key 不一定相同,这段代码作为结构参考,不要求直接运行。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 下面是 LangChain RAG 的典型结构示例,作为项目结构参考\n", "\n", "# from langchain_text_splitters import RecursiveCharacterTextSplitter\n", "# from langchain_community.vectorstores import FAISS\n", "# from langchain_openai import OpenAIEmbeddings, ChatOpenAI\n", "# from langchain_core.prompts import ChatPromptTemplate\n", "\n", "# text_splitter = RecursiveCharacterTextSplitter(\n", "# chunk_size=500,\n", "# chunk_overlap=50,\n", "# )\n", "# docs = text_splitter.create_documents(documents)\n", "\n", "# embeddings = OpenAIEmbeddings(model=\"qwen3-embedding\")\n", "# vectorstore = FAISS.from_documents(docs, embeddings)\n", "# retriever = vectorstore.as_retriever(search_kwargs={\"k\": 3})\n", "\n", "# prompt = ChatPromptTemplate.from_template(\"\"\"\n", "# 请根据下面的资料回答用户问题。\n", "# 如果资料中没有答案,请说明不知道,不要编造。\n", "\n", "# 资料:\n", "# {context}\n", "\n", "# 问题:\n", "# {question}\n", "# \"\"\")\n", "\n", "# llm = ChatOpenAI(model=\"qwen3.6-35b-A3b\")\n", "# chain = prompt | llm\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码展示了真实 LangChain 项目的常见结构:\n", "\n", "1. `RecursiveCharacterTextSplitter` 用于文本切分,`chunk_size` 控制每块大小,`chunk_overlap` 控制相邻片段之间的重叠内容。\n", "2. `FAISS.from_documents(docs, embeddings)` 会把文档转换成向量并存入 FAISS 向量库。\n", "3. `vectorstore.as_retriever(search_kwargs={\"k\": 3})` 会创建检索器,每次返回最相关的 3 个片段。\n", "4. `ChatPromptTemplate.from_template(...)` 定义提示词模板,要求模型必须根据资料回答。\n", "5. `prompt | llm` 是 LangChain 表达式语言 LCEL 的写法,表示把提示词输出交给大模型。\n", "\n", "这段代码被注释掉,是因为它依赖真实模型、Embedding 服务和相关包。课程中先理解结构,后续再接入真实模型即可。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 12. 用 LangGraph 设计项目工作流\n", "\n", "如果项目流程变复杂,例如要增加问题改写、检索质量检查、答案审核,就可以使用 LangGraph 编排流程。\n", "\n", "一个基础工作流可以设计为:\n", "\n", "```text\n", "用户问题 -> 检索节点 -> 回答生成节点 -> 输出结果\n", "```\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from typing import List, Tuple, TypedDict\n", "\n", "\n", "class QAState(TypedDict):\n", " question: str\n", " chunks: List[str]\n", " retrieved: List[Tuple[int, str]]\n", " answer: str\n", "\n", "\n", "def retrieve_node(state: QAState):\n", " retrieved = retrieve(state[\"question\"], state[\"chunks\"], top_k=3)\n", " return {\"retrieved\": retrieved}\n", "\n", "\n", "def answer_node(state: QAState):\n", " answer = generate_answer(state[\"question\"], state[\"retrieved\"])\n", " return {\"answer\": answer}\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码先定义 LangGraph 工作流需要的状态和节点:\n", "\n", "1. `QAState` 表示问答流程中的共享数据。\n", "2. `question` 保存用户问题。\n", "3. `chunks` 保存知识库切分后的文本片段。\n", "4. `retrieved` 保存检索结果,每个结果包含分数和文本片段。\n", "5. `answer` 保存最终答案。\n", "6. `retrieve_node` 是检索节点,负责调用前面写好的 `retrieve` 函数。\n", "7. `answer_node` 是回答节点,负责调用 `generate_answer` 生成答案。\n", "\n", "可以看到,LangGraph 节点并不神秘,本质上就是接收 State、处理数据、返回更新字段的普通函数。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 13. 构建 LangGraph 工作流\n", "\n", "下面把检索节点和回答节点连接起来。\n", "\n", "如果当前环境没有安装 LangGraph,可以先运行:`%pip install langgraph`。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# 如果没有安装 LangGraph,取消下一行注释后运行\n", "# %pip install langgraph\n", "\n", "from langgraph.graph import END, START, StateGraph\n", "\n", "\n", "qa_workflow = StateGraph(QAState)\n", "\n", "qa_workflow.add_node(\"retrieve\", retrieve_node)\n", "qa_workflow.add_node(\"answer\", answer_node)\n", "\n", "qa_workflow.add_edge(START, \"retrieve\")\n", "qa_workflow.add_edge(\"retrieve\", \"answer\")\n", "qa_workflow.add_edge(\"answer\", END)\n", "\n", "qa_app = qa_workflow.compile()\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码把问答流程编排成 LangGraph 工作流:\n", "\n", "1. `StateGraph(QAState)` 创建一张基于 `QAState` 的流程图。\n", "2. `add_node(\"retrieve\", retrieve_node)` 添加检索节点。\n", "3. `add_node(\"answer\", answer_node)` 添加回答生成节点。\n", "4. `add_edge(START, \"retrieve\")` 表示从检索节点开始。\n", "5. `add_edge(\"retrieve\", \"answer\")` 表示检索完成后进入回答节点。\n", "6. `add_edge(\"answer\", END)` 表示回答生成后流程结束。\n", "7. `compile()` 把流程图编译成可以调用的 `qa_app`。\n", "\n", "这就是一个最小 LangGraph 版个人知识库问答流程。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 14. 运行 LangGraph 问答助手\n", "\n", "现在传入问题和知识库片段,运行整个工作流。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "initial_state = {\n", " \"question\": \"个人知识库问答助手有什么作用?\",\n", " \"chunks\": all_chunks,\n", " \"retrieved\": [],\n", " \"answer\": \"\",\n", "}\n", "\n", "final_state = qa_app.invoke(initial_state)\n", "print(final_state[\"answer\"])\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 代码解释\n", "\n", "这段代码运行了 LangGraph 工作流:\n", "\n", "1. `initial_state` 是初始状态,包含用户问题、知识库片段、空的检索结果和空答案。\n", "2. `qa_app.invoke(initial_state)` 启动工作流。\n", "3. 数据会先进入 `retrieve_node`,生成 `retrieved`。\n", "4. 然后进入 `answer_node`,生成 `answer`。\n", "5. 最终返回 `final_state`,我们打印其中的 `answer`。\n", "\n", "这个例子说明:LangGraph 适合把多个处理步骤组织成清晰、可维护的项目流程。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 15. 项目可以如何继续升级\n", "\n", "当前版本只是教学版 MVP,后续可以从这些方向升级:\n", "\n", "1. 支持读取 PDF、Markdown、Word 等真实文件。\n", "2. 使用真实 Embedding 模型,把关键词检索升级为向量检索。\n", "3. 使用 Chroma 或 FAISS 保存向量索引。\n", "4. 接入真实大模型,生成更自然的回答。\n", "5. 增加“无法回答”判断,避免资料不足时强行回答。\n", "6. 增加来源展示,让用户看到答案引用了哪些资料。\n", "7. 使用 LangGraph 增加问题改写、答案审核、多轮追问等节点。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 16. 小结\n", "\n", "本节课完成了个人知识库问答助手的第一版项目设计和实现。\n", "\n", "你需要重点掌握:\n", "\n", "1. 项目开发要先做需求分析,再写代码。\n", "2. RAG 的核心流程是:准备资料、切分资料、检索资料、生成答案。\n", "3. LangChain 更适合提供文档加载、切分、检索、模型调用等组件。\n", "4. LangGraph 更适合把多个步骤编排成清晰工作流。\n", "5. 初学项目时,可以先用纯 Python 理解原理,再逐步替换成成熟框架。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 17. 练习题\n", "\n", "1. 往 `documents` 中新增 3 条自己的学习笔记,然后重新运行问答流程。\n", "2. 修改 `top_k`,观察返回 1 条、3 条、5 条资料时答案有什么变化。\n", "3. 修改 `simple_score`,让它按词语而不是按字符计算相关度。\n", "4. 在 `generate_answer` 中增加“参考资料”部分,把检索到的片段编号展示出来。\n", "5. 给 LangGraph 工作流新增一个 `check_node`,判断是否检索到了相关资料。\n" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.14.4" } }, "nbformat": 4, "nbformat_minor": 4 }