LangChain 学习路线 · 前端工程师的 8 周计划

前端工程师 · 8 周 · Python

从组件树到检索链

给前端工程师的 8 周 Python LangChain 实战路线。不堆概念，按周推进；每个知识点配可运行案例，RAG 从加载到生成逐环节拆解，最后接上你最熟的那一层：前端。

8 周 · 7 个 RAG 环节 · Python 3.11 · LangChain 1.x · 约 45 分钟读完

RAG 全链路示意

阶段	流程
离线索引	1 文档加载 → 2 切分 → 3 向量化 → 4 入库
在线查询	5 检索 → 6 重排 → 7 生成 + 引用

---

01 8 周路线图

每天 1~1.5 小时：30 分钟读文档，60 分钟写代码，15 分钟记笔记。不要只看不写——每周都必须留下一个能跑的东西。

周	主题	产出
0	Python 速成（前端视角）	能写异步脚本、装 venv、读写 pydantic 模型
1	LLM 调用 + Prompt + LCEL	命令行聊天 + 流式输出
2	结构化输出、工具调用、消息历史	会「调函数」的助手
3	RAG 上半场：加载 → 切分 → 向量化 → 入库	本地知识库可检索
4	RAG 下半场：检索 → 重排 → 生成 → 引用	带出处的问答 demo
5	高级检索：混合检索、多查询、ParentDoc、HyDE	检索命中率明显提升
6	评估 + LangSmith + FastAPI 上线	有指标的 API 服务
7	LangGraph + Agent（含 Agentic RAG）	自适应 RAG 图
8	结课项目 + 前端联调	可演示作品

先建立心智模型

LangChain 里的大部分概念，都能在浏览器里找到对应物。先映射，再学 API，速度会快很多。

LangChain	前端对应物
ChatPromptTemplate	模板字符串 / JSX props
LCEL a | b	RxJS pipe() / Promise 链 / Unix 管道
RunnableParallel	Promise.all
chain.stream()	SSE / ReadableStream
Document	带 metadata 的对象
VectorStore / Retriever	Algolia / ES 索引，封装好的 search()
LangGraph / checkpointer	状态机 + Redux reducer / 本地存档
LangSmith	Sentry + Chrome DevTools
FastAPI / LangServe	BFF 层

版本提醒

网上大量教程是 0.0.x / 0.1，API 差异很大。1.x 里经典链（create_retrieval_chain 等）迁到了 langchain-classic，新项目推荐 LCEL 手动组装 或 LangGraph。下文的 RAG 主代码用手动 LCEL，版本兼容性最好。

---

02 Python 速成 · 第 0 周 · 3~5 天

只学用得到的：环境、类型、异步、文件与 JSON。会 TypeScript 的话，这一周主要是换语法。

环境

python -m venv .venv
.venv\Scripts\activate
pip install -U langchain langchain-openai langchain-community `
  langchain-text-splitters langchain-chroma python-dotenv

requirements.txt 一定要锁版本，LangChain 迭代很快：

langchain==1.*
langchain-openai==1.*

类型 + pydantic（对应 TS interface + zod）

from pydantic import BaseModel, Field

class Query(BaseModel):
    text: str
    top_k: int = Field(default=4, ge=1, le=20)  # zod 的 min/max

q = Query(text="年假怎么算", top_k=3)
print(q.model_dump())   # {'text': '年假怎么算', 'top_k': 3}

异步（对应 async/await + Promise.all）

import asyncio

async def fetch(i: int) -> str:
    await asyncio.sleep(0.5)
    return f"result-{i}"

async def main():
    results = await asyncio.gather(*(fetch(i) for i in range(5)))
    print(results)

asyncio.run(main())

必须掌握

list / dict 推导式、生成器 yield、f-string、try/except、with、*args / **kwargs、json、pathlib、从 .env 读取配置。

---

03 LangChain 核心抽象 · 第 1–2 周

这一章的目标不是背 API，而是理解一条数据怎么从字符串流进模型、再流回前端。

最小心智模型

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    # 国内模型走 OpenAI 兼容协议，改 base_url 即可：
    # base_url="https://api.deepseek.com/v1", model="deepseek-chat"
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}，回答简洁。"),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()   # 这就是 LCEL
print(chain.invoke({"role": "老师", "question": "什么是向量数据库？"}))

要点：invoke 传 dict → 模板填充 → 模型 → 解析。| 左边输出自动成为右边输入。

流式（前端最熟）

for chunk in chain.stream({"role": "老师", "question": "讲个冷笑话"}):
    print(chunk, end="", flush=True)

并行 + 透传

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

chain = RunnableParallel(
    original=RunnablePassthrough(),        # 原样保留输入
    upper=lambda x: x["q"].upper(),
)
chain.invoke({"q": "hello"})
# {'original': {'q': 'hello'}, 'upper': 'HELLO'}

结构化输出（对应 zod parse）

from pydantic import BaseModel

class Movie(BaseModel):
    title: str
    year: int

structured = llm.with_structured_output(Movie)
print(structured.invoke("推荐一部 2010 年后的科幻电影"))
# Movie(title='星际穿越', year=2014)

工具调用（Agent 的地基）

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询城市天气。"""
    return f"{city} 晴，25℃"

llm_with_tools = llm.bind_tools([get_weather])
resp = llm_with_tools.invoke("北京天气")
print(resp.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, ...}]

多轮历史

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是助手"),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])
base = prompt | llm | StrOutputParser()

_store = {}
def get_history(sid: str):
    return _store.setdefault(sid, InMemoryChatMessageHistory())

chat = RunnableWithMessageHistory(
    base, get_history,
    input_messages_key="question", history_messages_key="history",
)
chat.invoke({"question": "我叫小明"}, config={"configurable": {"session_id": "u1"}})

稳健性三件套

chain = chain.with_retry(stop_after_attempt=3).with_fallbacks([backup_chain])
config = {"tags": ["prod"], "metadata": {"user_id": "u1"}}

给前端的提示

把 Runnable 当成 Observable：可以组合、可以 stream、可以 batch、可以 with_retry。LCEL 的 | 就是 pipe。

---

04 RAG 全链路 · 第 3–4 周

RAG = 离线索引（加载 → 切分 → 向量化 → 入库）+ 在线查询（检索 → 重排 → 拼 Prompt → 生成 → 引用）。下面七个环节，每个都给实现和坑。

阶段	做什么	失败会怎样
索引	把文档变成可检索的向量条目	检索不到、检索到噪声
查询	把问题变成答案，并标注出处	答非所问、幻觉、无法追溯

1. 文档加载（Loading）

原理：所有来源统一成 Document(page_content, metadata)。metadata 是后面过滤、引用、去重的命脉。

from langchain_community.document_loaders import (
    PyPDFLoader, WebBaseLoader, TextLoader, DirectoryLoader,
)

# 单个 PDF：按页生成 Document，metadata 自带 source/page
docs = PyPDFLoader("handbook.pdf").load()

# 目录批量
docs = DirectoryLoader(
    "./docs", glob="**/*.txt",
    loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"},
).load()

# 网页
docs = WebBaseLoader(["https://example.com/a", "https://example.com/b"]).load()

for d in docs[:3]:
    print(d.metadata, d.page_content[:80])

坑

• 扫描版 PDF 需要 OCR（UnstructuredPDFLoader + OCR 依赖），否则全是空文本。
• 必须补齐业务 metadata：source / page / dept / updated_at / permission。
• 大批量用 loader.lazy_load() 省内存。

2. 切分（Splitting）

原理：embedding 有长度限制，检索粒度也要小于语义单元。分块质量决定 RAG 上限。

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter,
)

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,        # 中文按“字符”算，约 400~600 token 量级
    chunk_overlap=120,     # 10%~15%
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
)
chunks = splitter.split_documents(docs)
print(len(chunks), chunks[0].metadata)

Markdown 保留标题层级（标题写进 metadata，检索时更好定位）：

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
)
md_chunks = md_splitter.split_text(markdown_text)

坑

• chunk_size 太小 → 语义断裂；太大 → 噪声多、召回不准。用真实问题调参。
• 表格、代码块被切断 → 先按结构切，再按长度切。
• 中文别用英文默认分隔符，会切出整段超长块。

3. 向量化（Embedding）

原理：双塔模型把文本映射为向量，语义近则余弦相似度高。索引和查询必须用同一个 embedding 模型。

from langchain_openai import OpenAIEmbeddings

emb = OpenAIEmbeddings(model="text-embedding-3-small", dimensions=512)

# 本地/免费方案（需 pip install langchain-huggingface sentence-transformers）
# from langchain_huggingface import HuggingFaceEmbeddings
# emb = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")

开发期加缓存，别反复花钱：

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore

cached_emb = CacheBackedEmbeddings.from_bytes_store(
    emb, LocalFileStore("./emb_cache"), namespace=emb.model
)

坑

• 换 embedding 模型 = 必须全量重建索引，维度不同会直接报错。
• 部分模型要求 query 加指令前缀（如 BGE 系列），索引与查询写法要对齐。

4. 入库（VectorStore）

from langchain_chroma import Chroma

vectorstore = Chroma(
    collection_name="handbook",
    embedding_function=cached_emb,
    persist_directory="./chroma_db",
)

vectorstore.add_documents(chunks)          # 内部自动向量化并落盘
print(vectorstore.similarity_search_with_score("年假多少天", k=4))

其他选型：FAISS（纯本地、快）、PGVector（要事务/权限/多租户）、Milvus / Qdrant（大规模）。

坑

• 重复导入要固定 ids 做 upsert；删文档要同步 delete(ids)。
• 生产别把“全量重建”当更新策略。

5. 检索（Retrieval）

# 基础 TopK
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# MMR：兼顾相关性和多样性，避免 4 条都是重复内容
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

# 相似度阈值
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.3, "k": 4},
)

# metadata 过滤（多租户/权限/部门，非常关键）
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 4, "filter": {"dept": "hr"}}
)

坑

• k 太小漏召回，太大淹没有效信息。
• filter 字段要建索引；问题与文档术语不一致时，先做查询改写。

6. 后处理 / 重排（Rerank）

原理：向量召回是双塔（快但粗），rerank 是 cross-encoder（慢但准）。召回 20 条 → 重排取 4 条。

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank   # 国内可用 BGE-reranker / DashScope gte-rerank

reranker = CohereRerank(model="rerank-multilingual-v3.0", top_n=4)
rerank_retriever = ContextualCompressionRetriever(
    base_compressor=reranker, base_retriever=retriever
)

长上下文重排（把最相关的放头尾，利用“中间遗忘”）：

from langchain_community.document_transformers import LongContextReorder

reordered = LongContextReorder().transform_documents(docs)

坑

rerank 是延迟大头。先量化收益，再决定是否上；top_n 不要比喂给模型的条数还大。

7. 生成 + 引用（Generation）

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

def format_docs(docs):
    return "\n\n".join(
        f"[{i+1}] (来源:{d.metadata.get('source')} p.{d.metadata.get('page')})\n{d.page_content}"
        for i, d in enumerate(docs)
    )

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是企业知识库助手。只依据给定资料回答，资料不足就说“根据现有资料无法确定”。"
     "回答后用 [1][2] 标注引用片段。"),
    ("human", "资料：\n{context}\n\n问题：{question}"),
])

rag = (
    RunnableParallel(
        context=rerank_retriever | format_docs,
        question=RunnablePassthrough(),
        docs=rerank_retriever,          # 同时把原始文档带出来做引用
    )
    | prompt | llm | StrOutputParser()
)

answer = ""
for chunk in rag.stream("年假怎么算？"):
    answer += chunk
    print(chunk, end="", flush=True)

多轮 RAG 要先把追问改写成独立问题，否则“那它呢？”检索不到：

condense = (
    ChatPromptTemplate.from_messages([
        ("system", "根据历史把追问改写成独立完整的问题，只输出问题。"),
        ("human", "历史:\n{history}\n追问:{question}"),
    ]) | llm | StrOutputParser()
)

坑

• 不约束“只依据资料”就必然幻觉。
• 引用编号必须和 format_docs 的编号一致。
• context 超长要裁剪或分级（stuff → map_reduce）。

经典写法

旧教程常见 create_stuff_documents_chain + create_retrieval_chain，1.x 中在 langchain-classic。能用，但上面的手动 LCEL 更透明，也更容易加引用和流式。

---

05 高级检索 · 第 5 周

基础 RAG 跑通后，提分通常来自这里。按收益排序：混合检索 → 重排 → 查询改写 → 父子文档。

混合检索（向量 + BM25）

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever  # 1.x: langchain_classic.retrievers

bm25 = BM25Retriever.from_documents(chunks, k=4)   # 中文需接 jieba 分词
vec = vectorstore.as_retriever(search_kwargs={"k": 4})
hybrid = EnsembleRetriever(retrievers=[bm25, vec], weights=[0.4, 0.6])

专有名词、型号、人名用 BM25 补足向量的短板。

多查询改写（Multi-Query）

from langchain.retrievers.multi_query import MultiQueryRetriever

mq = MultiQueryRetriever.from_llm(retriever=vec, llm=llm)

LLM 生成 3 个不同问法，并行检索后去重，提升召回。

HyDE（假设文档嵌入）

def hyde_search(q: str, k: int = 4):
    fake_answer = (llm | StrOutputParser()).invoke(f"用一段话回答：{q}")
    return vectorstore.similarity_search(fake_answer, k=k)

问题太短、太口语化时效果明显。

父子文档（Small-to-Big）

小块检索、大块喂模型：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
store = InMemoryStore()

pdr = ParentDocumentRetriever(
    vectorstore=vectorstore, docstore=store,
    child_splitter=child_splitter, parent_splitter=parent_splitter,
)
pdr.add_documents(docs)

Self-Query（自然语言转 metadata 过滤）

“2023 年 HR 部门的制度”自动拆出 filter={"dept":"hr","year":2023}，用 SelfQueryRetriever.from_llm + 字段描述实现。适合元数据规范的业务库。

Agentic RAG

检索 → 相关性打分 → 不合格就改写问题重查 → 合格才生成。这是 Self-RAG / CRAG 的核心思想，下一章用 LangGraph 实现。

---

06 评估与上线 · 第 6 周

没有评估的 RAG 优化全是玄学。先建 30~100 条黄金问答，再谈调参。

看哪些指标

侧	指标	回答什么问题
检索	Recall@k / Hit Rate / MRR	该找到的文档找到了吗
检索	Context Precision / Recall	召回的上下文准不准、全不全
生成	Faithfulness	答案是否忠于资料
生成	Answer Relevancy / Correctness	是否答到点上、与标准答案一致

RAGAS 案例

from datasets import Dataset
from ragas import evaluate
# metrics 在新版里是类，需要实例化
from ragas.metrics import Faithfulness, ResponseRelevancy, LLMContextRecall

data = Dataset.from_list([{
    "question": "年假多少天",
    "answer": rag.invoke("年假多少天"),
    "contexts": [d.page_content for d in rerank_retriever.invoke("年假多少天")],
    "ground_truth": "入职满一年 5 天，逐年递增",
}])
result = evaluate(data, metrics=[Faithfulness(), ResponseRelevancy(), LLMContextRecall()])
print(result)

每次改切分、embedding 或 rerank，都跑一遍。LangSmith 的 Datasets / Experiments 可以管理这套回归测试。

用 FastAPI 上线

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(q: str):
    async def gen():
        async for chunk in rag.astream(q):
            yield f"data: {json.dumps({'delta': chunk}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

工程化清单

• astream_events 做细粒度事件；.with_retry()、超时、限流、降级模型。
• LLM 与 embedding 双层缓存，成本大头常在 embedding 和 rerank。
• LangSmith 追踪：LANGSMITH_TRACING=true + LANGSMITH_API_KEY，用来定位“是检索错了还是生成错了”。
• 知识库增量 upsert + 删除；每天全量重建不可持续。

---

07 LangGraph 与 Agent · 第 7 周

当流程需要分支、循环、人工审批和断点续跑时，LCEL 就不够了。LangGraph 用“状态 + 节点 + 边”表达，像 Redux。

最小图

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

g = StateGraph(State)
g.add_node("chatbot", chatbot)
g.add_edge(START, "chatbot")
g.add_edge("chatbot", END)
app = g.compile()
print(app.invoke({"messages": [("user", "你好")]}))

自适应 RAG（条件边 + 循环）

class RAGState(TypedDict):
    question: str
    docs: list
    tries: int

def retrieve(s): return {"docs": retriever.invoke(s["question"])}

def grade(s):
    ok = bool(s["docs"])
    return "generate" if ok else "rewrite"

def rewrite(s):
    return {"question": s["question"] + " 详细说明", "tries": s["tries"] + 1}

g = StateGraph(RAGState)
g.add_node("retrieve", retrieve)
g.add_node("generate", lambda s: {"docs": s["docs"]})
g.add_node("rewrite", rewrite)
g.add_edge(START, "retrieve")
g.add_conditional_edges("retrieve", grade, {"generate": "generate", "rewrite": "rewrite"})
g.add_edge("rewrite", "retrieve")
g.add_edge("generate", END)
app = g.compile()

生产里 grade 用 LLM 做相关性判断，并加 tries 上限防止死循环。

存档与中断

from langgraph.checkpoint.memory import MemorySaver

app = g.compile(checkpointer=MemorySaver(), interrupt_before=["generate"])
app.invoke(init, config={"configurable": {"thread_id": "u1"}})

thread_id 相当于会话 ID；checkpoint 让多轮、人工审批、断点续跑成为可能。

一行建 Agent（1.x）

from langchain.agents import create_agent  # 旧版: langgraph.prebuilt.create_react_agent

agent = create_agent(model=llm, tools=[search_kb], system_prompt="先查知识库再回答")

把检索封装成工具，Agent 自己决定何时查、查几次。

---

08 项目实战 · 第 8 周

选 1~2 个做完，比看十个教程有用。

企业知识库问答（推荐）

PDF / Markdown 目录 → 切分入库 → 混合检索 + rerank → 流式回答 + 引用跳转 → RAGAS 报告 → FastAPI + 前端 SSE。

代码库问答

按函数 / 类切分，metadata 存 file:line，回答直接给代码链接。

带工具的客服 Agent

知识库检索 + 订单查询 API + 工单创建，LangGraph 控制流程与人工审批。

文档对比助手

两个版本 diff 后分块，检索时按 version 过滤，回答指出差异条款。

交付物

GitHub 仓库（README 写清架构图、指标、成本）、演示视频、一页“踩坑与优化记录”。这三样就是你的作品集。

---

09 前端联调 · 你的主场

后端输出 SSE，前端用 ReadableStream 解析。Key 只放服务端，前端只管渲染与引用浮层。

const res = await fetch("/chat", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ q: "年假怎么算" }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";

while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  const lines = buffer.split("\n\n");
  buffer = lines.pop() ?? "";
  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;
    const data = line.slice(6);
    if (data === "[DONE]") return;
    console.log(JSON.parse(data).delta);
  }
}

• 用 Next.js Route Handler 做 BFF 代理，密钥只放服务端。
• 引用数据单独走一个事件，前端渲染成可点击的来源列表。
• 记得处理断流、重连、取消（AbortController）和错误提示。

---

10 避坑清单

这些坑基本每个人都会踩一遍，提前看一眼能省两周。

1. 别从 Agent 开始。 先 LCEL，再 RAG，再 LangGraph。顺序反了会一直“调不通”。
2. 教程版本对不上时，先看 pip show langchain，再看官方文档当前版。
3. 中文切分按字符算，chunk_size=800 别照搬英文教程的 token 数。
4. 换 embedding 模型必须重建索引，先把维度写进配置。
5. 没有 metadata 的 RAG 是玩具：权限、部门、时间、来源都得进 metadata。
6. 不加“只依据资料回答”的约束，幻觉率会高得离谱。
7. 先建 30 条黄金评测集再调参，不然全是玄学。
8. 内存历史只能开发用，MemorySaver / InMemoryChatMessageHistory 生产要换 Redis / Postgres。
9. 成本大头常在 embedding 和 rerank，不在生成；开发期一定要开缓存。
10. Prompt 注入：检索到的文档也可能含恶意指令，系统提示要明确“资料只是数据”。
11. 更新知识库用增量 upsert + 删除，不要每天全量重建。
12. 上线前加超时、重试、降级，LLM 不是稳定依赖。

---

11 自检清单

能全部做到，就算出师。

• 能解释 LCEL 中 | 的数据流，能手写 RunnableParallel
• 能独立搭一条带引用和流式的 RAG
• 能说清 chunk_size / overlap / k / fetch_k / top_n 各自影响什么
• 能设计 metadata 方案支持权限过滤
• 能跑 RAGAS 并用分数指导优化
• 能用 LangGraph 写带条件边和 checkpoint 的流程
• 能把服务用 FastAPI 包装成 SSE 接口并对接前端
• 能用 LangSmith 定位“是检索错了还是生成错了”

---

按自己的节奏推进，每周至少留下一个可运行的 demo。

Python 3.11 · LangChain 1.x · RAG