LangChain本地实操七日速成：从环境搭建到LangGraph状态机

1. 项目概述：为什么“一周速成”不是画饼，而是可落地的本地实操路径

LangChain 这个词最近两年在技术圈里出现的频率，已经不亚于当年的 Docker 或者 React。但和那些工具不同，LangChain 不是一个装完就能跑的命令行程序，它更像一套“AI 应用的乐高积木说明书”——你得先理解每块积木的咬合逻辑、承重边界、拼接顺序，再动手搭出能跑起来的小车或吊桥。很多人卡在第一步：打开官网文档，满屏的from langchain_core.prompts import ChatPromptTemplate，配上一堆抽象概念（Chain、Agent、Tool、Retriever），没环境、没数据、没调试反馈，学三天就放弃。这根本不是学习能力问题，是路径错了。

我带过二十多个从零起步的工程师做 RAG 和 Agent 项目，发现一个铁律：所有卡点，90% 都发生在本地环境启动失败、依赖冲突、向量库写不进数据、或者提示词一改就崩这四个环节。而市面上绝大多数“LangChain 入门教程”，要么直接跳到云端托管服务（比如用 LangChain + LlamaIndex + Pinecone 三件套在线部署），要么堆砌概念图解，却没人告诉你 conda 环境里 pip install langchain==0.1.20 为什么会报ImportError: cannot import name 'BaseModel' from 'pydantic'，也没人提醒你 FAISS 在 M1 Mac 上默认编译失败时该加哪两个环境变量。这些不是“细节”，是拦在“能跑起来”和“完全懵圈”之间的那堵墙。

所以这个“一周速成计划”，核心不是教你背多少 API，而是用七天时间，每天聚焦一个可验证、可打断、可回溯的本地实操目标：

• 第1天：conda 创建纯净 Python 3.11 环境，用 pip 安装 LangChain 及其全部硬依赖（llama-index、faiss-cpu、openai、tiktoken），全程离线缓存复用，避免网络抖动导致安装中断；
• 第2天：不用任何外部模型，用本地 Ollama 加载 Qwen2:1.5b，跑通最简 Chain：输入一段文本 → 切分 → 嵌入 → 存入 FAISS → 检索 → 生成摘要；
• 第3天：把本地知识库换成 Markdown 文件夹，实现“投喂即检索”，重点解决中文分词不准、标题层级丢失、代码块污染向量的问题；
• 第4天：引入提示工程实战，对比 system prompt 写法对 RAG 输出稳定性的影响，用真实案例展示“请用三句话总结” vs “请按‘背景-方法-结论’结构输出”带来的结果差异；
• 第5天：升级为 Agent，用 LangChain 自带的 Tool + ReAct 框架，让 AI 能主动调用本地计算器、读取 CSV 表格、甚至执行 shell 命令（仅限安全沙箱）；
• 第6天：接入 MySQL 作为结构化数据源，实现“自然语言查数据库”，比如问“上个月销售额最高的三个产品是什么”，自动转成 SQL 并返回结果；
• 第7天：用 LangGraph 重构第5天的 Agent 流程，加入人工审核节点、失败重试分支、状态持久化（checkpoint-blob），跑通一个带记忆、可中断、可恢复的生产级对话流。

这个计划不承诺让你写出 LangChain 源码，但保证第七天结束时，你能独立完成：
✅ 本地一键初始化 LangChain 开发环境（Windows/Mac/Linux 全适配）；
✅ 把任意 PDF/Word/Markdown 文档变成可问答的知识库（RAG 实战）；
✅ 设计并验证至少3种不同场景下的提示词模板（Prompt Engineering）；
✅ 构建一个能调用本地工具、查数据库、带错误处理的多步 Agent；
✅ 看懂 LangGraph 的 StateGraph 图谱，并修改其中任意一个节点逻辑。

适合谁？不是给算法博士看的，而是给：

• 已会 Python 基础（能写函数、读文件、用 requests）、想快速落地 AI 应用的后端/全栈工程师；
• 数据分析师，想用自然语言直接查 BI 报表背后的数据库；
• 产品经理，需要亲手验证 RAG 对客服知识库的提效程度；
• 学生党，毕业设计要做“智能法律咨询系统”或“论文辅助阅读工具”。

它不讲“LangChain 是干嘛的”这种教科书定义，只讲“你现在手头有台笔记本，明天就要让 AI 读懂你硬盘里的 200 页产品手册”，该怎么一步步敲出第一行有效代码。

2. 核心思路拆解：为什么必须坚持“本地环境实操”，而不是跳过环境直奔模型

很多人看到“LangChain 一周速成”，第一反应是：“直接用 HuggingFace 的 Inference API 不香吗？何必折腾本地？” 这是个极典型的认知偏差——把 LangChain 当成了模型推理框架，而它本质是应用编排层（Orchestration Layer）。你可以把它理解成厨房里的“厨师长”：他不自己种菜（不训练模型）、不自己烧火（不管理 GPU 显存）、甚至不亲自切菜（不处理 tokenization），但他必须清楚：

• 哪个灶台（LLM）火力猛但费油（Qwen2-7B 推理快但显存占用高）；
• 哪把刀（Embedding Model）切肉丝最细（bge-m3 中文嵌入效果好但体积大）；
• 哪个冰柜（Vector DB）存菜最保鲜（FAISS 本地快但不支持分布式，Chroma 轻量但并发弱）；
• 哪张菜单（Prompt Template）能让客人（用户）点单不翻车（system prompt 写错会导致 RAG 忽略检索结果）。

而这一切判断，必须建立在本地可控、可调试、可复现的环境之上。举三个真实踩坑案例：

案例1：云端 API 的“黑盒延迟”毁掉 RAG 体验
某客户用 LangChain + OpenAI API 做客服知识库，测试时响应 800ms，上线后突增到 3.2s。排查发现是 OpenAI 的 rate limit 触发了排队，但 LangChain 默认重试策略是指数退避，用户连续提问两次就卡死。如果本地用 Ollama 跑 Qwen2:1.5b，你一眼就能看到ollama run qwen2:1.5b启动耗时、每次推理的 token/s、GPU 显存占用曲线——这些数据是云端 API 绝对不给你的。没有本地基线，你连“慢”都定义不了。

案例2：FAISS 的“ silently fail”陷阱
FAISS 在 Apple Silicon（M1/M2）芯片上，默认编译版本不启用 AVX2 指令集，导致向量检索精度暴跌 40%。很多教程直接pip install faiss-cpu，跑 demo 时“看起来能动”，但实际检索结果驴唇不对马嘴。而本地实操时，你运行python -c "import faiss; print(faiss.get_num_threads())"就能立刻发现线程数为 1（应为 CPU 核心数），进而意识到要重装faiss-cpu并设置export OMP_NUM_THREADS=8。这种底层硬件适配问题，只有亲手在终端敲命令才能暴露。

案例3：提示词工程的“幻觉放大器”效应
新手常犯的错误是：把 RAG 的 prompt 写成“请根据以下资料回答问题”，结果 LLM 直接忽略检索出的 chunk，凭空编造答案。本地实操时，你可以用langchain.debug=True打印完整 chain trace，清楚看到：

1. Retrieval 步骤返回了哪3个 chunk（含原文和 score）；
2. LLM 输入的完整 prompt 是什么（含 system + user + retrieved context）；
3. LLM 输出的 raw text 是什么。
   这种“所见即所得”的调试能力，是任何托管平台都无法提供的。你不是在猜模型怎么想，而是在看它每一步怎么算。

所以本计划的底层逻辑非常明确：

• Day 1–3 是“筑基”：用最笨的办法（conda + pip + 本地模型）把环境打穿，确保每个组件都能独立验证；
• Day 4–5 是“控场”：掌握 Prompt Engineering 和 Agent 编排，让 LangChain 不只是“能跑”，而是“跑得稳、答得准”；
• Day 6–7 是“升维”：接入结构化数据（MySQL）和状态机（LangGraph），把单次问答升级为可持续对话流。

这不是“速成捷径”，而是把别人绕开的坑，变成你亲手填平的路基。当你第七天用 LangGraph 写出StateGraph并成功保存 checkpoint-blob 到本地磁盘时，你会明白：所谓“速成”，不过是把模糊的“应该会”，变成了清晰的“我已经会”。

3. 核心细节解析与实操要点：从 conda 环境创建到 FAISS 向量入库的完整链路

3.1 Day 1：conda 环境创建与依赖安装——为什么必须用 conda 而非纯 pip？

很多人觉得“pip 就够了”，但在 LangChain 生态中，这是个高危操作。LangChain 依赖树极其复杂：

• langchain-core依赖pydantic>=2.5.0,<3.0.0；
• langchain-community依赖sqlalchemy>=2.0.0；
• faiss-cpu依赖numpy>=1.21.0且要求 ABI 兼容；
• ollamaPython SDK 依赖httpx>=0.23.0。

如果全用 pip，很容易触发“依赖地狱”：比如pip install langchain自动拉取最新版pydantic，但llama-index的某个旧版又强制要求pydantic<2.0，结果import llama_index直接报错。conda 的优势在于它用 SAT 求解器做依赖解析，能同时满足所有包的约束条件。

实操步骤（Windows/macOS/Linux 通用）：

1. 下载并安装 Miniconda（轻量版，比 Anaconda 小 80%）：

   # macOS curl -fsSL https://repo.anaconda.com/miniconda/Miniconda3-latest-MacOSX-arm64.sh -o miniconda.sh bash miniconda.sh -b -p $HOME/miniconda3 source $HOME/miniconda3/bin/activate # Windows 用户下载 Miniconda3 Windows 64-bit 安装包，勾选“Add Anaconda to my PATH”

2. 创建专用环境（关键！不要用 base 环境）：

   conda create -n langchain-dev python=3.11 conda activate langchain-dev

3. 利用 pip 本地缓存加速安装（解决国内网络不稳定问题）：

   提示：pip 默认缓存路径在~/.cache/pip（Linux/macOS）或%LOCALAPPDATA%\pip\Cache（Windows）。首次安装后，所有 wheel 包已存在本地，后续重装无需重复下载。

   # 先安装基础依赖（避免 langchain 自动拉取冲突版本） pip install numpy==1.26.4 pydantic==2.7.1 tiktoken==0.7.0 # 再安装 LangChain 全家桶（注意版本锁定，避免自动升级） pip install langchain==0.1.20 langchain-community==0.0.38 langchain-core==0.1.52 # 安装 FAISS（CPU 版，无 GPU 依赖） pip install faiss-cpu==1.8.0 # 安装 Ollama Python SDK（用于本地模型调用） pip install ollama==0.2.8

4. 验证环境是否干净：

   python -c "import langchain, faiss, ollama; print('✅ All imports successful')"

   如果报错ModuleNotFoundError: No module named 'faiss'，大概率是 FAISS 编译失败。此时执行：

   # macOS M1/M2 用户必加（否则 FAISS 向量计算精度归零） export OMP_NUM_THREADS=8 export OPENBLAS_NUM_THREADS=8 pip uninstall faiss-cpu -y && pip install faiss-cpu==1.8.0

注意：不要用pip install langchain[all]。这个命令会无差别安装所有可选依赖（包括 Redis、PostgreSQL、Docker SDK），而你当前只需要 FAISS 和 Ollama。过度安装不仅拖慢环境初始化，还会增加冲突概率。

3.2 Day 2：FAISS 向量库本地入库——从原始文本到可检索 chunk 的全流程

FAISS 是 Facebook 开源的高效相似性搜索库，LangChain 用它做向量检索的核心引擎。但很多人以为“装完 FAISS 就能搜”，其实中间隔着三道坎：文本切分（Splitting）、嵌入编码（Embedding）、向量存储（Indexing）。

Step 1：文本切分——为什么不能简单按句号分割？
中文文档里，“。”不一定代表语义断点。比如“Python 3.11 发布了。它性能提升 10%。” 这两句话共享主语“Python 3.11”，强行切分会导致检索时上下文丢失。LangChain 提供RecursiveCharacterTextSplitter，它按[\\n\\n, \\n, ' ', '']优先级递归切分，保留段落完整性。

from langchain.text_splitter import RecursiveCharacterTextSplitter # 针对中文优化的切分器 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每个 chunk 最多 500 字符 chunk_overlap=50, # 相邻 chunk 重叠 50 字符，避免语义割裂 separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""] # 中文标点优先 )

Step 2：嵌入模型选择——为什么 bge-m3 比 sentence-transformers 更适合中文 RAG？
sentence-transformers的paraphrase-multilingual-MiniLM-L12-v2是老牌选择，但中文语义捕获较弱。BAAI/bge-m3是 2024 年新发布的多向量嵌入模型，支持 dense、sparse、colbert 三种模式，对中文长尾词（如“微服务熔断机制”）召回率高 22%。它可通过HuggingFaceEmbeddings直接加载：

from langchain.embeddings import HuggingFaceEmbeddings embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-m3", model_kwargs={'device': 'cpu'}, # 本地 CPU 运行 encode_kwargs={'normalize_embeddings': True} )

实操心得：首次运行embeddings.embed_query("hello")会自动下载 2.3GB 模型权重。建议提前用huggingface-cli download BAAI/bge-m3 --local-dir ./bge-m3预下载，避免运行时卡住。

Step 3：FAISS 向量入库——为什么FAISS.from_documents()比手动add()更可靠？
手动调用index.add()需要自己处理向量维度、归一化、ID 映射，极易出错。FAISS.from_documents()封装了全部细节：

from langchain_community.vectorstores import FAISS from langchain_community.document_loaders import TextLoader # 加载本地文本（例如 README.md） loader = TextLoader("./data/README.md") docs = loader.load() # 切分 + 嵌入 + 入库 三步合一 vectorstore = FAISS.from_documents( documents=text_splitter.split_documents(docs), embedding=embeddings ) # 保存到本地磁盘（下次直接加载，无需重算） vectorstore.save_local("./faiss_index")

关键参数解释：

• chunk_size=500：不是越大越好。实测超过 800 字符，bge-m3 嵌入向量的 cosine similarity 分布会变宽，导致检索结果相关性下降；
• chunk_overlap=50：重叠太少（如 10）会导致跨段落信息丢失，太多（如 100）则增加冗余计算；
• normalize_embeddings=True：强制向量单位化，FAISS 的内积搜索等价于余弦相似度，精度更高。

注意：FAISS 默认使用IndexFlatIP（内积索引），对小规模数据（<10 万向量）足够快。如果未来扩展到百万级，需切换为IndexIVFFlat并训练聚类中心，但 Day 2 完全不需要。

3.3 Day 3：知识库“投喂”实战——如何让 LangChain 正确解析 Markdown 和 PDF

Day 2 用了纯文本，但真实知识库是 Markdown、PDF、甚至 Word。不同格式的 loader 行为差异极大，必须针对性处理。

Markdown 解析陷阱：
UnstructuredMarkdownLoader会把# 标题、- 列表、```python代码块全部当正文处理，导致向量污染。正确做法是用MarkdownHeaderTextExtractor提取标题层级，作为元数据注入：

from langchain_community.document_loaders import UnstructuredMarkdownLoader from langchain_community.document_transformers import MarkdownHeaderTransformer # 提取 #、##、### 标题作为 metadata headers_to_extract = [ ("#", "header_1"), ("##", "header_2"), ("###", "header_3"), ] loader = UnstructuredMarkdownLoader("./docs/api_guide.md") raw_docs = loader.load() transformer = MarkdownHeaderTransformer(headers_to_extract=headers_to_extract) docs = transformer.transform_documents(raw_docs) # 此时每个 doc.metadata 包含 'header_1': '认证流程', 'header_2': 'Token 有效期'

PDF 解析雷区：
PyPDFLoader只能提取文字，遇到扫描版 PDF（图片）直接返回空字符串。UnstructuredPDFLoader支持 OCR，但默认不启用。必须显式传参：

from langchain_community.document_loaders import UnstructuredPDFLoader loader = UnstructuredPDFLoader( "./docs/manual.pdf", mode="elements", # 按段落/标题/表格分类 strategy="hi_res", # 启用 OCR（需提前安装 unstructured[local-inference]） # 注意：hi_res 模式需额外下载 1.2GB 模型，首次运行会卡住 ) docs = loader.load()

实操心得：PDF OCR 非常耗时。建议先用strategy="fast"快速预览，确认是文字版 PDF 后再切hi_res。如果是扫描版，直接换pdfplumber提取文本+坐标，再用layoutparser识别标题区域——但这已超出 Day 3 范围，属于进阶技巧。

统一入库技巧：
不同 loader 返回的Document对象字段不一致（如PyPDFLoader有page_number，UnstructuredMarkdownLoader有category）。入库前必须标准化元数据：

for doc in docs: # 统一添加 source 文件名 doc.metadata["source"] = doc.metadata.get("source", "unknown") # 清理冗余字段（避免向量库膨胀） doc.metadata = {k: v for k, v in doc.metadata.items() if k in ["source", "header_1", "page_number"]}

这一步看似琐碎，但决定了后续 RAG 检索时能否按“来源文档”或“章节标题”精准过滤。没有元数据标准化，你的知识库就是一锅粥。

4. 实操过程与核心环节实现：从 RAG 到 Agent 的七日完整代码链

4.1 Day 4：提示工程实战——用三组对比实验，看清 system prompt 的真实威力

很多人以为提示词就是“多写几句话”，但 LangChain 的 RAG 流程中，prompt 是控制信息流向的阀门。我们用同一份知识库（Kubernetes 部署文档），设计三组实验：

代码实现（使用 LangChain 的 ChatPromptTemplate）：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser # 实验 C 的 prompt（推荐生产环境使用） prompt_c = ChatPromptTemplate.from_messages([ ("system", """你是一个 Kubernetes 专家。请严格按以下结构回答： 1. **定义**：用一句话说明该概念 2. **配置项**：列出所有必需参数及默认值（**必须从提供的文档中直接提取，不可编造**） 3. **示例**：给出最小可行 YAML 文档内容： {context}"""), ("human", "{question}") ]) # 绑定 LLM（Ollama 本地模型） from langchain_community.chat_models import ChatOllama llm = ChatOllama(model="qwen2:1.5b", temperature=0.1) # 构建 RAG chain rag_chain = ( {"context": vectorstore.as_retriever(), "question": lambda x: x["question"]} | prompt_c | llm | StrOutputParser() ) # 执行查询 result = rag_chain.invoke({"question": "如何配置 Pod 的健康检查？"}) print(result)

关键洞察：

• System prompt 中的指令强度（如“必须” vs “请”）直接影响 LLM 的服从度；
• 结构化要求（如“分三点回答”）能显著降低幻觉，但需配合具体格式描述（如“定义：”加粗）；
• 禁止性指令（如“不可编造”）比正面指令（如“请准确”）更有效，因为 LLM 对否定词更敏感。

实操心得：不要在 prompt 里写“如果你不知道，请说不知道”。LLM 会把这句话当废话忽略。正确做法是：在 retriever 层设置search_kwargs={"k": 3}，确保至少返回 3 个相关 chunk；再在 prompt 里写“仅基于以上文档内容回答，未提及的内容一律不回答”。这才是真正可控的“拒答”机制。

4.2 Day 5：Agent 实战——用 ReAct 框架让 AI 主动调用本地工具

LangChain 的 Agent 不是“更聪明的 LLM”，而是“会查手册的实习生”。它通过 ReAct（Reasoning + Acting）循环，自主决定何时调用工具、何时生成答案。

构建一个“本地文件分析 Agent”：

• Tool 1：ShellTool—— 执行ls -l查看目录；
• Tool 2：ReadFileTool—— 读取指定文件内容；
• Tool 3：CalculatorTool—— 计算数学表达式。

from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import ShellTool, ReadFileTool, CalculatorTool from langchain import hub # 加载 ReAct 提示模板（LangChain 官方维护） prompt = hub.pull("hwchase17/react") # 定义工具（限制权限，仅允许读取 ./data/ 目录） tools = [ ShellTool(allowed_commands=["ls"]), # 禁用 rm/cp 等危险命令 ReadFileTool(root_dir="./data/"), # 只能读 data 目录 CalculatorTool() # 安全计算 ] # 创建 Agent agent = create_react_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 执行任务 result = agent_executor.invoke({ "input": "查看 ./data/ 目录下有哪些文件？把 report.csv 的前 5 行内容读出来，并计算 123*456 的结果" }) print(result["output"])

Agent 的执行日志（关键观察点）：

> Entering new AgentExecutor chain... Thought: 我需要先查看 ./data/ 目录内容，使用 ShellTool。 Action: ShellTool Action Input: ls -l ./data/ Observation: total 8 -rw-r--r-- 1 user staff 1024 May 10 10:00 report.csv -rw-r--r-- 1 user staff 512 May 10 10:01 config.json Thought: 目录中有 report.csv，下一步读取其前 5 行。 Action: ReadFileTool Action Input: report.csv Observation: id,name,score 1,Alice,95 2,Bob,87 3,Charlie,92 4,Diana,88 5,Eve,96 Thought: 现在计算 123*456。 Action: CalculatorTool Action Input: 123 * 456 Observation: 56088 Thought: 我已获取所有必要信息，可以回答问题。 Final Answer: ./data/ 目录下有 report.csv 和 config.json。report.csv 前 5 行是... 计算结果是 56088。

为什么 ReAct 比传统 Chain 更可靠？

• 可追溯：每一步 Action 和 Observation 都记录，便于 debug；
• 自修正：如果ReadFileTool返回空，Agent 会自动重试或换工具；
• 免硬编码：不用写if "csv" in query: read_csv()这样的 if-else，LLM 自主决策。

注意：Agent 的verbose=True是调试生命线。生产环境关闭它，但开发阶段必须开着，否则你永远不知道 AI 为什么“突然不干活了”。

4.3 Day 6：接入 MySQL——让自然语言直接查数据库

RAG 处理非结构化文本，但业务数据在 MySQL 里。LangChain 的SQLDatabaseToolkit能把 NL 问题转成 SQL，前提是数据库 schema 清晰。

前提：准备 MySQL 环境

-- 创建示例数据库 CREATE DATABASE sales_db CHARACTER SET utf8mb4; USE sales_db; -- 创建订单表 CREATE TABLE orders ( id INT PRIMARY KEY AUTO_INCREMENT, product_name VARCHAR(100), amount DECIMAL(10,2), order_date DATE ); -- 插入测试数据 INSERT INTO orders VALUES (1, 'iPhone 15', 999.00, '2024-05-01'), (2, 'MacBook Pro', 1999.00, '2024-05-02'), (3, 'AirPods', 199.00, '2024-05-03');

LangChain 连接与查询：

from langchain.sql_database import SQLDatabase from langchain.agents.agent_toolkits import SQLDatabaseToolkit from langchain.agents import create_sql_agent # 连接数据库（注意：密码明文仅用于本地开发） db = SQLDatabase.from_uri("mysql+pymysql://root:password@localhost:3306/sales_db") # 创建 SQL Toolkit（自动分析表结构，生成列注释） toolkit = SQLDatabaseToolkit(db=db, llm=llm) # 创建 SQL Agent agent_executor = create_sql_agent( llm=llm, toolkit=toolkit, agent_type="openai-tools", # 使用 OpenAI 工具调用格式，兼容本地 LLM verbose=True ) # 执行自然语言查询 result = agent_executor.invoke({ "input": "上个月销售额最高的三个产品是什么？" }) print(result["output"])

Agent 自动生成的 SQL（关键验证点）：

SELECT product_name, SUM(amount) as total_sales FROM orders WHERE order_date >= '2024-04-01' AND order_date < '2024-05-01' GROUP BY product_name ORDER BY total_sales DESC LIMIT 3;

避坑指南：

• Schema 注释必须写：在 MySQL 中给表和列加 COMMENT，如COMMENT '产品名称'，否则 LLM 无法理解字段含义；
• 日期函数要显式声明：last_month这种模糊词，LLM 可能转成DATE_SUB(NOW(), INTERVAL 1 MONTH)或BETWEEN，需在 prompt 中指定格式；
• 敏感数据脱敏：生产环境必须用SQLDatabase的include_tables参数，只暴露必要表，禁用information_schema。

4.4 Day 7：LangGraph 升级——用状态机实现可中断、可恢复的对话流

LangChain 的 Agent 是“一次性的”，而 LangGraph 是“有记忆的”。它用StateGraph定义节点（Node）和边（Edge），让对话流变成可编程的状态机。

目标：构建一个“技术支持对话助手”，支持：

• 用户提问 → 检索知识库 → 生成答案；
• 若答案置信度低（score < 0.6），转入人工审核节点；
• 人工审核后，可选择“采纳”或“拒绝”，并更新向量库。

from typing import TypedDict, Annotated, List, Optional from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 定义状态结构 class GraphState(TypedDict): question: str context: str answer: str confidence: float needs_review: bool review_action: Optional[str] # "approve" or "reject" # 定义节点函数 def retrieve_node(state: GraphState) -> GraphState: # 检索知识库 docs = vectorstore.similarity_search_with_score(state["question"], k=3) state["context"] = "\n\n".join([doc[0].page_content for doc in docs]) state["confidence"] = min([doc[1] for doc in docs]) # 取最高分 state["needs_review"] = state["confidence"] < 0.6 return state def generate_node(state: GraphState) -> GraphState: # 生成答案 result = rag_chain.invoke({ "question": state["question"], "context": state["context"] }) state["answer"] = result return state def human_review_node(state: GraphState) -> GraphState: # 模拟人工审核（实际可对接 Web UI） print(f"⚠️ 人工审核：问题 '{state['question']}'，答案置信度 {state['confidence']:.2f}") print(f" 建议答案：{state['answer'][:100]}...") action = input("请输入 'approve' 或 'reject': ").strip() state["review_action"] = action return state # 构建图 workflow = StateGraph(GraphState) # 添加节点 workflow.add_node("retrieve", retrieve_node) workflow.add_node("generate", generate_node) workflow.add_node("review", human_review_node) # 添加边 workflow.add_edge(START, "retrieve") workflow.add_edge("retrieve", "generate") workflow.add_conditional_edges( "generate", lambda x: x["needs_review"], {True: "review", False: END} ) workflow.add_edge("review", END) # 设置内存检查点（关键！让对话可恢复） checkpointer = MemorySaver() app = workflow.compile(checkpointer=checkpointer) # 运行对话（支持中断后恢复） config = {"configurable": {"thread_id": "support_001"}} result = app.invoke({"question": "K8s 的 livenessProbe 如何配置？"}, config=config) print("最终答案：", result["answer"])

LangGraph 的核心价值：

• checkpointer：自动保存每步 state 到内存，关机重启后app.get_state(config)可恢复；
• conditional_edges：用 Python 函数动态决定流向，比硬编码 if-else 更灵活；
• thread_id：隔离不同用户的对话状态，避免交叉污染。

实操心得：LangGraph 的invoke()默认是同步阻塞的。如需异步支持（如 Websocket 推送进度），用app.astream()，它会 yield 每个节点的输出，前端可实时渲染“正在检索...”、“正在生成...”。

5. 常见问题与排查技巧实录：从 conda 环境冲突到 LangGraph 状态丢失的全链路排障

5.1 环境类问题：