Day26 LangChain 框架开发

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. LangChain 到底解决什么问题

Day25 把 RAG 全链路手写了一遍：文档解析、切片、向量化、灌库、检索、重排、上下文组装、Prompt 拼接、调用大模型、解析输出。这些步骤里，真正体现业务差异的只有「切片策略」「检索策略」「Prompt」三处，其余全是胶水代码。

LangChain 的定位就是把这些胶水代码抽象成统一接口 + 可替换实现：

没有框架时：                        用 LangChain 后：
                                  
读文件  → 自己写 fs/pdf-parse        DocumentLoader（统一 load() → Document[]）
切片    → 自己写递归切分逻辑          TextSplitter（统一 splitDocuments()）
向量化  → 自己调 embedding HTTP       Embeddings（统一 embedQuery/embedDocuments）
灌库    → 自己写 Milvus SDK 调用      VectorStore（统一 addDocuments/similaritySearch）
检索    → 自己写 topK + 过滤          Retriever（统一 invoke(query) → Document[]）
拼 Prompt→ 自己写模板字符串           PromptTemplate（统一变量插值 + 消息角色）
调模型  → 自己写 OpenAI/通义 HTTP     ChatModel（统一 invoke/stream/batch）
解析输出→ 自己写 JSON.parse + 兜底    OutputParser（统一 parse，带容错）
串起来  → 自己写一堆 await            LCEL（用 | 把上面全部串成一条 Runnable）

一句话概括：LangChain = LLM 应用的「标准接口层」+ 一套编排 DSL（LCEL）。它不提供模型、不提供向量库，只提供「让你换模型、换库时不用改业务代码」的抽象。

面试加分表述：LangChain 的价值不在于"帮我少写代码"，而在于"让我的业务代码不依赖具体供应商"。真正省下来的不是行数，而是切换 OpenAI → 通义、Milvus → PGVector 时的改造成本。

2. 八大核心组件与它们的输入输出契约

组件	职责	输入 → 输出	典型实现
Models	调用大模型	`string \| BaseMessage[]` → `AIMessage`	`ChatOpenAI`、`ChatTongyi`、`ChatOllama`
Prompts	模板化拼装输入	`Record<string, any>` → `PromptValue`	`PromptTemplate`、`ChatPromptTemplate`
DocumentLoaders	把各种源读成文档	路径/URL → `Document[]`	`PDFLoader`、`CheerioWebBaseLoader`
TextSplitters	长文切片	`Document[]` → `Document[]`	`RecursiveCharacterTextSplitter`
Embeddings	文本转向量	`string[]` → `number[][]`	`OpenAIEmbeddings`、`AlibabaTongyiEmbeddings`
VectorStores	存储与相似检索	`Document[]` → ✓ / `query` → `Document[]`	`MemoryVectorStore`、`Milvus`、`PGVectorStore`
Retrievers	统一检索抽象	`string` → `Document[]`	`VectorStoreRetriever`、`MultiQueryRetriever`
Chains / Runnable	编排	任意 → 任意	LCEL 组合、`createRetrievalChain`
Memory / History	多轮上下文	`sessionId` → `BaseMessage[]`	`RunnableWithMessageHistory`

Document 是贯穿全链路的统一数据结构，只有两个字段：

interface Document {
  pageContent: string                  // 正文
  metadata: Record<string, any>        // 来源、页码、切片序号、业务字段
}

⚠️ metadata 是 RAG 工程化的命脉。Day25 讲的引用溯源、增量灌库的 docId 版本号、权限过滤的 tenantId，全部挂在 metadata 上。LangChain 的所有 Loader/Splitter 都会自动透传 metadata，这是它比自己写胶水代码省事的最大一处。

3. 新版包结构：为什么 `import` 路径和教程不一样

LangChain 从 0.1 开始做了包拆分，这是面试和踩坑的高频区：

@langchain/core          ← 只有抽象基类与 LCEL（Runnable、BaseMessage、Document）
     ↑ 被依赖                 零三方依赖，版本最稳定
     │
     ├── @langchain/openai      ← OpenAI provider（ChatOpenAI、OpenAIEmbeddings）
     ├── @langchain/anthropic   ← Claude provider
     ├── @langchain/community   ← 社区实现（各种 Loader / VectorStore / 国产模型）
     │
     └── langchain              ← 上层封装（预制 Chain、Agent、Retriever 组合）

版本阶段	导入方式	问题
0.0.x（旧教程）	`import { ChatOpenAI } from "langchain/chat_models/openai"`	单包巨大，装一个模型把所有依赖都拉下来
0.1+（现在）	`import { ChatOpenAI } from "@langchain/openai"`	按需安装，核心抽象独立

迁移要点：

业务代码只依赖 @langchain/core 的类型，provider 包仅在工厂函数里出现，这样换模型时改一个文件。
peerDependencies 必须把 @langchain/core 锁成同一版本，否则会出现两份 Runnable 基类，导致 | 组合时报 "not a Runnable"。
自己写的工具库要把 @langchain/core 声明为 peer 而不是 dependency。

4. Models：LLM 与 ChatModel 的差异

// LLM：纯文本进、纯文本出（已基本被淘汰）
const res1 = await llm.invoke("你好")            // => "你好！有什么可以帮你"

// ChatModel：消息数组进、AIMessage 出（现在的事实标准）
const res2 = await chat.invoke([
  new SystemMessage("你是严谨的技术助手"),
  new HumanMessage("你好"),
])
// => AIMessage { content: "...", response_metadata: { tokenUsage, finish_reason } }

四类消息角色：

消息类型	用途	注意
`SystemMessage`	角色设定、全局约束	多数模型只认第一条，中途插入可能被忽略
`HumanMessage`	用户输入	多模态时 content 可为数组（文本 + 图片 URL）
`AIMessage`	模型回复	带 `tool_calls` 时表示要调工具
`ToolMessage`	工具执行结果	必须带 `tool_call_id` 与上一条 AIMessage 对应

统一的调用参数（换 provider 不用改）：

const chat = new ChatOpenAI({
  modelName: "gpt-4o-mini",
  temperature: 0,          // RAG 场景必须 0，保证可复现
  maxTokens: 2048,
  timeout: 30_000,         // 必配，否则默认无超时会把请求挂死
  maxRetries: 2,           // 内置指数退避重试
  streaming: true,
})

面试加分表述：RAG 的生成环节 temperature 必须设 0 或接近 0。因为答案应当完全由检索到的上下文决定，采样随机性在这里只会制造幻觉，不会带来"创造力"。

5. Prompts：模板、消息模板与 FewShot

// 1. 字符串模板
const p1 = PromptTemplate.fromTemplate("把下面内容翻译成{lang}：\n{text}")
await p1.format({ lang: "英文", text: "你好" })

// 2. 聊天模板（推荐，天然支持角色）
const p2 = ChatPromptTemplate.fromMessages([
  ["system", "你是客服助手，只能依据<context>回答，无依据就说不知道。"],
  new MessagesPlaceholder("history"),   // 历史消息占位，接多轮
  ["human", "<context>\n{context}\n</context>\n\n问题：{question}"],
])

MessagesPlaceholder 是接入多轮对话的关键：它接收的是 BaseMessage[] 而不是字符串，不会被模板引擎转义。

部分变量（partial）用于固化不变量，避免每次调用都传：

const prompt = await p2.partial({ today: new Date().toISOString().slice(0, 10) })

⚠️ 模板里的 {} 是插值语法。如果 Prompt 里要写 JSON 示例，必须写成 {{ 和 }} 转义，否则直接抛 "Missing value for input"。这是最常见的新手报错。

6. DocumentLoaders：把异构数据源归一成 Document

const loaders = {
  pdf:  new PDFLoader("./a.pdf", { splitPages: true }),
  docx: new DocxLoader("./a.docx"),
  csv:  new CSVLoader("./a.csv", { column: "content" }),
  web:  new CheerioWebBaseLoader(url, { selector: "article" }),
  dir:  new DirectoryLoader("./docs", { ".md": p => new TextLoader(p) }),
}
const docs = await loaders.pdf.load()   // Document[]，metadata 自带 source/loc.pageNumber

能力	说明
`load()`	一次性全读进内存，小文件用
`loadAndSplit(splitter)`	边读边切，省内存但无法做跨页合并
流式 `lazyLoad()`	大文件必用，逐个 yield Document

工程经验：Loader 只解决"读得出来"，不解决"读得对"。PDF 的双栏排版、表格、页眉页脚，官方 Loader 一律原样拼接。生产环境通常是「自研解析（OCR/版面分析）→ 手动 new Document()」，Loader 仅用于 demo 和简单格式。这与 Day25 的结论一致：解析环节必须自研。

7. TextSplitters：递归切分的分隔符优先级

RecursiveCharacterTextSplitter 是默认选择，核心是按语义强度从强到弱递归尝试：

默认分隔符顺序：["\n\n", "\n", " ", ""]

切分逻辑：
  用 "\n\n" 切 → 每块是否 ≤ chunkSize？
      ├── 是 → 收下
      └── 否 → 对这块用 "\n" 再切 → 还超？
                  └── 用 " " 切 → 还超？
                          └── 用 "" 逐字符硬切（兜底）

const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 500,
  chunkOverlap: 80,                       // 重叠，防止答案被切断在边界
  separators: ["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],  // 中文必须自定义
})
const chunks = await splitter.splitDocuments(docs)

问题	原因	解法
中文被切得稀碎	默认分隔符没有中文标点，`" "` 在中文里几乎不存在，直接掉到逐字符硬切	显式加入 `。！？；，`
Markdown 标题和正文被切开	递归切分不认结构	用 `MarkdownTextSplitter` 或 `fromLanguage("markdown")`
代码被切断	同上	`RecursiveCharacterTextSplitter.fromLanguage("js")`，内置语言级分隔符
overlap 设太大	召回重复内容，浪费上下文	经验值 `chunkSize` 的 10%~20%

⚠️ chunkSize 的单位默认是字符数而非 token 数。中文 1 字约 0.6~1 token，英文 1 token 约 4 字符。要精确控 token 必须传 lengthFunction 用 tokenizer 计算，否则上下文长度估算会整体偏差。

8. VectorStores 与 Retrievers 的分层关系

VectorStore（存储层）                Retriever（检索策略层）
  ├── addDocuments()                   └── invoke(query) → Document[]
  ├── similaritySearch(q, k)
  ├── similaritySearchWithScore()         Retriever 只有一个契约方法，
  ├── delete({ ids })                     所以任何检索逻辑都能伪装成 Retriever：
  └── asRetriever({ k, filter })          ES / 关键词 / 图数据库 / HTTP 接口

const store = await MemoryVectorStore.fromDocuments(chunks, embeddings)

// 带过滤与打分阈值的检索器
const retriever = store.asRetriever({
  k: 8,
  filter: { tenantId: "t_001", status: "published" },   // 权限 + 状态过滤
  searchType: "mmr",            // 最大边际相关，降低召回冗余
  searchKwargs: { fetchK: 30, lambda: 0.6 },
})

asRetriever() 这层抽象的意义：上游 Chain 只认 Retriever 接口，所以把 MemoryVectorStore 换成 Milvus 时，业务代码一行不改。

面试加分表述：Retriever 是 LangChain 里设计最克制、价值最高的抽象——只有 invoke(query) → Document[] 一个方法。我们生产环境的混合检索（向量 + ES + 业务规则重排）完全是自研的，但只要实现 Retriever 接口，就能无缝塞进 LangChain 的链路里，框架和自研代码的边界非常清晰。

9. Chains 的演进：从 LLMChain 到 LCEL

// ❌ 旧写法（0.0.x，已 deprecated）
const chain = new LLMChain({ llm, prompt })
const res = await chain.call({ question })      // 返回 { text: "..." }，key 还得猜

// ✅ 新写法（LCEL）
const chain = prompt.pipe(llm).pipe(new StringOutputParser())
const res = await chain.invoke({ question })    // 直接返回 string

旧 Chain 被废弃的根本原因：

旧 Chain 的问题	LCEL 如何解决
输入输出是 `Record<string, any>`，key 靠约定，无类型	泛型化 `Runnable<I, O>`，输入输出类型可推导
不支持流式，只能等全部生成完	任意组合自动获得 `stream()`
并行要自己写 `Promise.all`	`RunnableParallel` 声明式并行
重试/降级要自己包 try-catch	`.withRetry()` / `.withFallbacks()`
中间步骤不可观测	统一 Callback + LangSmith 自动追踪

10. LCEL：管道符的本质

LCEL（LangChain Expression Language）不是新语法，而是所有组件都实现了同一个 Runnable 接口，pipe() 只是把前一个的输出喂给后一个：

// JS 里用 .pipe()，Python 里用 | 运算符（重载了 __or__）
const chain = RunnableSequence.from([prompt, llm, parser])
// 等价于
const chain = prompt.pipe(llm).pipe(parser)

Runnable 的五个核心方法，组合后自动继承：

方法	语义	场景
`invoke(input)`	单次同步调用	普通接口
`batch(inputs[])`	批量并发调用	离线评测、批量打标
`stream(input)`	流式产出	打字机效果
`streamEvents(input)`	产出链路每一步的事件	前端展示"正在检索/正在生成"
`streamLog()`	产出 JSON Patch 形式的增量状态	调试、进度条

面试加分表述：LCEL 的核心价值是"组合即继承能力"。我只写了 prompt.pipe(llm).pipe(parser) 三个节点的串联，但这条链自动具备了流式、批量、异步、可观测、可重试五种能力——这些如果自己写胶水代码，每条链路都要重复实现一遍。

11. 三个最常用的 Runnable 原语

// RunnablePassthrough：原样透传，常用于"保留原始输入"
// RunnableParallel（对象字面量即可）：并行执行，结果合并成对象
// RunnableLambda（函数即可）：把普通函数变成 Runnable

const ragChain = RunnableSequence.from([
  {
    context: async (input) => {
      const docs = await retriever.invoke(input.question)
      return docs.map((d, i) => `[${i + 1}] ${d.pageContent}`).join("\n\n")
    },
    question: (input) => input.question,        // 普通函数自动包成 RunnableLambda
    history: (input) => input.history ?? [],
  },                                            // 这个对象 = RunnableParallel，三路并发
  prompt,
  llm,
  new StringOutputParser(),
])

await ragChain.invoke({ question: "怎么申请报销？" })

关键点：对象字面量会被自动转成 RunnableParallel，里面每个 value 都是 Runnable 或函数，它们接收同一份输入、并发执行、结果按 key 合并后传给下一节点。这是 LCEL 最常用的模式——「扇出取数 → 汇聚 → 渲染 Prompt」。

12. OutputParsers：从字符串到结构化

Parser	输出	说明
`StringOutputParser`	`string`	取 `AIMessage.content`，最常用
`JsonOutputParser`	`object`	容错解析，支持流式增量 JSON
`StructuredOutputParser`	按 schema 的对象	会生成格式说明注入 Prompt
`CommaSeparatedListOutputParser`	`string[]`	轻量列表

// ✅ 优先用模型原生结构化输出，比 Parser 可靠得多
const structured = llm.withStructuredOutput(z.object({
  answer: z.string(),
  citations: z.array(z.number()),
  refused: z.boolean(),
}))
const res = await prompt.pipe(structured).invoke({ context, question })
// res 已是类型安全的对象，底层走 function calling / json_schema，不靠模型"听话"

⚠️ StructuredOutputParser 的原理是往 Prompt 里塞格式说明，然后硬解析模型输出，模型不听话就报错。而 withStructuredOutput() 走的是模型 API 的 response_format / tools 能力，由服务端保证 JSON 合法。能用后者就不要用前者。

13. Memory 的现状：旧 API 已废弃

// ❌ 旧写法（BufferMemory + ConversationChain），0.3 起 deprecated
const memory = new BufferMemory({ memoryKey: "history", returnMessages: true })
const chain = new ConversationChain({ llm, memory })

// ✅ 新写法：RunnableWithMessageHistory
const store = new Map<string, InMemoryChatMessageHistory>()
const withHistory = new RunnableWithMessageHistory({
  runnable: chain,
  getMessageHistory: async (sessionId) => {
    if (!store.has(sessionId)) store.set(sessionId, new InMemoryChatMessageHistory())
    return store.get(sessionId)!
  },
  inputMessagesKey: "question",
  historyMessagesKey: "history",
})

await withHistory.invoke({ question: "它的上限是多少？" }, {
  configurable: { sessionId: "u_1001" },       // sessionId 从 config 传，不是从 input
})

三种历史管理策略对比：

策略	实现	优点	代价
全量缓冲	直接拼全部历史	信息无损	token 线性上涨，轮数一多必超限
滑动窗口	只保留最近 N 轮	成本恒定	早期信息丢失
摘要压缩	旧历史用 LLM 压成摘要	成本可控且保留主线	多一次 LLM 调用，摘要本身可能失真

生产组合拳：最近 3 轮原文 + 更早历史的滚动摘要 + 硬性 token 上限截断，三者同时生效。

⚠️ InMemoryChatMessageHistory 只能单进程用。多实例部署必须换成 RedisChatMessageHistory / UpstashRedisChatMessageHistory，否则负载均衡一打到别的实例，历史就"失忆"了。

14. 预制 Chain：`createRetrievalChain` 全家桶

// 1. 文档塞进 Prompt 的链（stuff 策略）
const combineDocsChain = await createStuffDocumentsChain({ llm, prompt })

// 2. 带历史感知的 Query 改写检索器（解决多轮指代）
const historyAwareRetriever = await createHistoryAwareRetriever({
  llm, retriever,
  rephrasePrompt: ChatPromptTemplate.fromMessages([
    new MessagesPlaceholder("chat_history"),
    ["human", "{input}"],
    ["human", "基于以上对话，把最后这个问题改写成不依赖上下文的独立问题，只输出问题本身。"],
  ]),
})

// 3. 组装成完整 RAG 链
const chain = await createRetrievalChain({
  retriever: historyAwareRetriever,
  combineDocsChain,
})
const res = await chain.invoke({ input: "它支持哪些格式？", chat_history })
// res = { input, chat_history, context: Document[], answer: string }

createHistoryAwareRetriever 正好对应 Day25 讲的 Query 改写：多轮里用户说"它"、"那个"，直接拿去向量检索必然召回不准，必须先用 LLM 补全成独立问题。

四种文档合并策略：

策略	做法	适用
`stuff`	全部塞进一个 Prompt	文档少、上下文够用，默认选它
`map_reduce`	每篇单独总结再汇总	文档多且可独立处理，并发快但丢关联
`refine`	顺序迭代精炼答案	需要渐进整合，串行慢
`map_rerank`	每篇出答案 + 置信度，取最高	答案只可能在单篇里

15. Python 版与 JS 版：同一套设计，两套手感

LangChain 有 Python 和 JS/TS 两个独立实现，设计理念一致（都是 Runnable + LCEL），但 API 命名和生态成熟度有差异。本手册 Day22～Day25 的服务端是 FastAPI + Python，本日为了前端同学读起来顺手用 TS 示范；面试里被问到的几乎都是 Python 生态，所以必须能双向翻译。

python

# Python：同一条 RAG 链，注意 | 运算符与 snake_case
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(retriever.invoke(x["question"])))
    | ChatPromptTemplate.from_template("依据资料回答：\n{context}\n问题：{question}")
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 同步 / 异步 / 流式 / 批量：四套方法名
chain.invoke({"question": "..."})            # 同步
await chain.ainvoke({"question": "..."})     # 异步（FastAPI 里必须用这个）
async for chunk in chain.astream({...}): ... # 流式
chain.batch([{...}, {...}])                  # 批量

能力	Python	JS / TS
核心包	`langchain-core`	`@langchain/core`
模型包	`langchain-openai`	`@langchain/openai`
社区集成	`langchain-community`	`@langchain/community`
管道组合	`a \| b \| c`（重载 `__or__`）	`a.pipe(b).pipe(c)`
命名风格	`snake_case`	`camelCase`
异步方法	另一套 `a` 前缀：`ainvoke` / `astream` / `abatch`	方法本身就是 `Promise`，无需区分
事件流	`astream_events(input, version="v2")`	`streamEvents(input, { version: "v2" })`
结构化输出	`with_structured_output(Model)`（Pydantic）	`withStructuredOutput(schema)`（Zod）
容错	`with_retry()` / `with_fallbacks([...])`	`withRetry()` / `withFallbacks([...])`
工具定义	`@tool` 装饰器 + Pydantic 入参	`tool(fn, { schema })` + Zod
token 统计	`get_openai_callback()` 上下文管理器	无，需自己写 Callback 累加
Loader / Splitter 覆盖度	最全（`unstructured`、各类办公文档、数据库）	明显少，复杂文档解析常要自己写或走 Python 侧
LangGraph 成熟度	功能领先（checkpoint、human-in-the-loop 更完整）	跟进中
部署形态优势	配合 FastAPI / Celery 做重计算	配合 Edge / Serverless 做纯编排与流式转发

⚠️ 三个只在 Python 侧踩的坑： ① astream_events 不传 version="v2" 会走到旧实现或直接报错，v1 的事件名和结构都不同； ② 在 FastAPI 的 async def 里调同步 invoke，会阻塞事件循环，并发一上来整个服务都卡 —— 异步接口里必须用 ainvoke / astream； ③ 自定义函数或 tool 里要透传 config（def f(x, config: RunnableConfig)），否则 callback、trace、configurable 会在这一层断掉，LangSmith 上看到的链路会断成两截。

面试加分表述：两端是「同一份接口规范的两个实现」，不是同一个库的两种语言绑定 —— 所以版本节奏、社区集成数量都不一样。我们的分工是重数据处理（文档解析、离线灌库、评测）放 Python，面向用户的编排与流式转发放 Node，中间用统一的检索服务接口对齐，避免同一套业务逻辑在两端各写一遍。

✨ 高级深挖知识点

1. Runnable 接口全貌：一张表吃透 LCEL

分类	方法 / 原语	作用
执行	`invoke` / `batch` / `stream` / `streamEvents`	四种执行语义，组合后自动继承
组合	`pipe()`、`RunnableSequence.from()`	串行
并行	对象字面量、`RunnableParallel`	扇出并发
透传	`RunnablePassthrough`、`.assign()`	保留原输入 / 在原输入上追加字段
函数化	`RunnableLambda`、普通函数	任意同步/异步函数接入链路
分支	`RunnableBranch`	if/else 路由
动态	`RunnableLambda` 返回 Runnable	运行时决定下一步用哪条链
容错	`.withFallbacks([...])`	主链失败自动切备用链
重试	`.withRetry({ stopAfterAttempt })`	节点级指数退避
配置	`.withConfig()`、`.configurableFields()`	运行时改 temperature / model
绑定	`.bind({ stop, tools })`	固化调用参数

.assign() 是最容易被忽略但最好用的一个：

// 在保留原输入的同时追加字段，避免手写 { ...input, context }
const chain = RunnablePassthrough.assign({
  context: async (i) => formatDocs(await retriever.invoke(i.question)),
}).pipe(prompt).pipe(llm).pipe(new StringOutputParser())

await chain.invoke({ question: "报销上限", userId: "u1" })
// prompt 能同时拿到 question / userId / context

2. RunnableBranch：按意图路由到不同链路

真实客服系统不会只有一条 RAG 链。闲聊、查订单、知识问答应当走不同处理：

const classify = ChatPromptTemplate
  .fromTemplate("把问题分类成 chat / order / kb 之一，只输出单词：{question}")
  .pipe(llm).pipe(new StringOutputParser())

const router = RunnableBranch.from([
  [(i) => i.intent === "order", orderChain],
  [(i) => i.intent === "kb",    ragChain],
  chatChain,                                   // 最后一个是 default
])

const app = RunnablePassthrough
  .assign({ intent: (i) => classify.invoke({ question: i.question }) })
  .pipe(router)

面试加分表述：用分类模型做路由，比把所有能力塞进一个超大 Prompt 效果好得多。超大 Prompt 会让指令互相干扰、token 成本恒定偏高；而路由后每条链的 Prompt 都很聚焦，成本和准确率同时受益。代价是多一次分类调用，可以用小模型或本地规则兜底来抵消。

3. 流式：`stream` 与 `streamEvents` 的本质区别

// stream：只拿最终节点的增量输出（用户看到的打字机）
for await (const chunk of chain.stream({ question })) process.stdout.write(chunk)

// streamEvents：拿链路每一步的生命周期事件（前端进度提示）
for await (const ev of chain.streamEvents({ question }, { version: "v2" })) {
  switch (ev.event) {
    case "on_retriever_end": send({ type: "docs", data: ev.data.output.length }); break
    case "on_chat_model_stream": send({ type: "token", data: ev.data.chunk.content }); break
    case "on_chain_end": send({ type: "done" }); break
  }
}

维度	`stream`	`streamEvents`
产出	末端输出的增量块	全链路事件（start/stream/end/error）
粒度	token	节点级 + token 级
典型用途	打字机	"正在检索知识库…已找到 5 篇…正在生成"
成本	低	事件量大，需按 `ev.name` 过滤

⚠️ 链路中间有非流式节点，整条链就会退化成非流式。比如中间插了一个 RunnableLambda 做全量 JSON 解析，它必须拿到完整输入才能算，前面的流就被"堵"住了。要保持流式，中间节点要么是透传，要么实现 transform()。

4. 流式 + 引用：先给引用还是先给正文

RAG 流式有个经典矛盾：引用列表要等答案生成完才知道用了哪几篇，但用户希望立刻看到内容。两种方案：

方案	做法	体验
先发召回文档	检索完立刻把 `Document[]` 的标题/链接推给前端，正文随后流	用户先看到"来源"，可信度高，推荐
后发真实引用	正文流完，再根据 `[1][2]` 标记回填真实引用	引用更准，但有等待感

生产做法是两者结合：检索完先推候选来源列表（灰色），正文流完后把实际被引用的高亮、未引用的收起。

5. 容错：`withFallbacks` + `withRetry` 的正确层级

const primary = new ChatOpenAI({ modelName: "gpt-4o", timeout: 20_000, maxRetries: 0 })
const backup  = new ChatOpenAI({ modelName: "gpt-4o-mini", timeout: 20_000 })

const llmWithGuard = primary
  .withRetry({ stopAfterAttempt: 2 })        // 先对同一模型重试（应对瞬时 5xx）
  .withFallbacks([backup])                   // 重试仍失败才降级到备用模型

错误类型	该重试吗	该降级吗
429 限流	✅ 带退避重试	✅ 重试无效则切备用 key/模型
5xx / 网络抖动	✅	✅
400 参数错误	❌ 重试必然还是错	❌ 直接抛
上下文超长	❌	✅ 降级到长上下文模型，或截断上下文重试
内容审核拒绝	❌	❌ 返回合规话术

⚠️ 配置 withRetry 时记得把模型自身的 maxRetries 设为 0，否则两层重试相乘——maxRetries:2 × stopAfterAttempt:3 = 最多 6 次真实请求，限流场景下会雪上加霜。

6. 自定义 Retriever：把自研检索塞进框架

import { BaseRetriever, type BaseRetrieverInput } from "@langchain/core/retrievers"
import type { Document } from "@langchain/core/documents"

class HybridRetriever extends BaseRetriever {
  lc_namespace = ["app", "retrievers"]
  constructor(private deps: Deps, fields?: BaseRetrieverInput) { super(fields) }

  // 只需实现这一个方法，整个框架就认你
  async _getRelevantDocuments(query: string): Promise<Document[]> {
    const [vec, kw] = await Promise.all([
      this.deps.vectorStore.similaritySearchWithScore(query, 20),
      this.deps.es.search(query, 20),
    ])
    const fused = rrfFuse(vec, kw)                    // Day25 讲的 RRF 融合
    const reranked = await this.deps.reranker.rerank(query, fused, 6)
    return reranked.filter(d => d.score >= 0.35)      // 打分阈值过滤，为拒答留出口
  }
}

关键点：

实现的是带下划线的 _getRelevantDocuments，外部调 invoke()；基类在外层包了 Callback 与追踪，不要直接重写 invoke，否则链路追踪会断。
lc_namespace 用于序列化标识，自定义类随便给个数组即可。
返回 Document[]，把分数写进 metadata.score，供上游拒答判断。

面试加分表述：我们团队的原则是「检索自研、编排用框架」。混合检索、重排、权限过滤全是自研的，但统一包装成 Retriever 接口接入 LCEL，这样既保留了对召回质量的完全控制，又能免费拿到流式、追踪、重试这些框架能力。

7. 官方高级 Retriever：什么时候真的该用

Retriever	原理	适用 / 代价
`MultiQueryRetriever`	LLM 把一个问题改写成 N 个，分别检索后去重合并	提升召回率；代价是多 N 次检索 + 1 次 LLM
`ParentDocumentRetriever`	小块建索引（检索准），命中后返回大块（上下文全）	性价比最高，需额外 docstore 存父文档
`SelfQueryRetriever`	LLM 把自然语言解析成「语义查询 + 结构化过滤条件」	适合"2024 年之后的财务制度"这类带条件的问法；对模型稳定性依赖高
`ContextualCompressionRetriever`	召回后用 LLM / 重排模型裁掉无关句子	省上下文、提忠实度；多一次模型调用
`EnsembleRetriever`	多个 Retriever 加权 RRF 融合	向量 + BM25 的标准组合

ParentDocumentRetriever 解决的是 Day25 的核心矛盾——切片越小检索越准，但上下文越碎：

索引层：  [小块 200 字] [小块 200 字] [小块 200 字]   ← 向量检索打这一层
                 ↓ 都指向
返回层：  [父文档 2000 字]                            ← 实际喂给 LLM 的是这个

8. Tools 与 Agent：让模型自己决定调什么

import { tool } from "@langchain/core/tools"
import { z } from "zod"

const queryOrder = tool(
  async ({ orderId }) => JSON.stringify(await orderApi.get(orderId)),
  {
    name: "query_order",
    description: "根据订单号查询订单状态与物流信息。只有用户明确给出订单号时才调用。",
    schema: z.object({ orderId: z.string().describe("订单号，纯数字，18 位") }),
  },
)

const agent = await createToolCallingAgent({ llm, tools: [queryOrder, kbSearch], prompt })
const executor = new AgentExecutor({
  agent, tools: [queryOrder, kbSearch],
  maxIterations: 5,              // 必须限制，否则可能无限循环烧 token
  returnIntermediateSteps: true, // 要调试就打开
})

Agent 的循环本质：

用户输入
   ↓
LLM 判断 → 不需要工具 → 直接输出最终答案 ✓
   ↓ 需要工具
返回 AIMessage.tool_calls = [{ name, args, id }]
   ↓
AgentExecutor 执行对应 tool
   ↓
把结果包成 ToolMessage（带 tool_call_id）追加到消息列表
   ↓
再次调 LLM ──────────────────┘（循环，直到无 tool_calls 或超过 maxIterations）

⚠️ description 就是 Agent 的"源码"。模型完全依赖 name + description + schema 的 describe() 来决策，描述写得模糊，调用准确率立刻崩。务必写清「什么时候该调、什么时候不该调、参数从哪来」。

9. Agent 的真实代价与 LangGraph

问题	表现	缓解
不可控	同一问题两次走不同路径	温度设 0；能用固定链路就不要用 Agent
成本高	每轮循环都是一次完整 LLM 调用	限制 `maxIterations`，工具数量控制在 10 个内
难调试	中间步骤黑盒	`returnIntermediateSteps` + LangSmith
无法人工介入	高危操作（退款、删除）直接被模型执行	加人工确认节点
状态难管理	复杂多步任务状态散落	改用 LangGraph

LangGraph 是 LangChain 官方对 Agent 的重写：把 Agent 从「隐式 while 循环」变成显式状态图，支持循环、分支、检查点（可持久化中断/恢复）、human-in-the-loop。结论：

固定流程（RAG 问答）→ LCEL 链，确定、便宜、好测。
需要动态决策但步骤可枚举 → LangGraph，可控可恢复。
完全开放的探索任务 → AgentExecutor，但生产慎用。

10. Callback：可观测性的统一入口

class MetricsHandler extends BaseCallbackHandler {
  name = "metrics"
  private t0 = new Map<string, number>()

  handleLLMStart(_llm, _p, runId) { this.t0.set(runId, Date.now()) }

  handleLLMEnd(output, runId) {
    const usage = output.llmOutput?.tokenUsage
    metrics.timing("llm.latency", Date.now() - this.t0.get(runId)!)
    metrics.count("llm.tokens.prompt", usage?.promptTokens ?? 0)
    metrics.count("llm.tokens.completion", usage?.completionTokens ?? 0)
    this.t0.delete(runId)
  }

  handleRetrieverEnd(docs, runId) { metrics.count("retriever.hits", docs.length) }
  handleChainError(err, runId) { logger.error({ runId, err }, "chain failed") }
}

// 两种挂载方式
const chain2 = chain.withConfig({ callbacks: [new MetricsHandler()] })   // 构造期，全局生效
await chain.invoke(input, { callbacks: [new MetricsHandler()] })         // 调用期，单次生效

常用钩子：handleLLMStart/NewToken/End/Error、handleChainStart/End/Error、handleToolStart/End、handleRetrieverStart/End。

runId / parentRunId 构成一棵调用树，把业务 traceId 通过 metadata 透传下去，就能和 APM 打通：

await chain.invoke(input, { metadata: { traceId, userId, tenantId }, tags: ["rag", "v2"] })

⚠️ Callback 默认同步阻塞执行。在 handleLLMEnd 里写数据库会直接拖慢响应，必须改成投递到队列异步处理。

11. LangSmith：链路追踪落地

bash

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxx
LANGCHAIN_PROJECT=kb-assistant-prod

只要这三个环境变量在，所有 Runnable 的执行自动上报，无需改一行代码。能看到：

每个节点的输入输出全文、耗时火焰图、token 与费用；
失败节点的完整堆栈与入参，可一键复现；
把线上 case 收集成 Dataset，跑回归评测（对应 Day25 的三层评测）。

数据合规是落地的最大障碍：LangSmith 是 SaaS，会把 Prompt 和文档内容上传。企业内网方案：

方案	说明
自建 LangSmith（企业版）	官方支持私有化，但要钱
自定义 Callback 上报	把 run 树写进自家 ES/ClickHouse，自己做 UI
OpenTelemetry	用 OTel 的 GenAI 语义约定上报到现有 APM，推荐，复用公司监控体系
脱敏后上报	Callback 里对 Prompt/文档做掩码再传

12. 成本与缓存

// 1. token 统计（Node 侧靠 callback 累加 llmOutput.tokenUsage；Python 有 get_openai_callback）
// 2. LLM 结果缓存：相同 Prompt 直接命中，省钱省延迟
import { setLLMCache } from "langchain/globals"
setLLMCache(new RedisCache({ client, ttl: 3600 }))

// 3. Embedding 缓存：同一段文本重复灌库时不重复付费
const cachedEmbeddings = CacheBackedEmbeddings.fromBytesStore(
  new OpenAIEmbeddings(), new RedisByteStore({ client }), { namespace: "text-embedding-3-small" },
)

缓存层	命中条件	注意
LLM 精确缓存	Prompt 字符串完全一致	Prompt 里有时间戳就永远不命中；temperature>0 时缓存会抹掉多样性
语义缓存	问题向量相似度超阈值	阈值定高了没效果，定低了会答非所问，必须带人工校验
Embedding 缓存	文本 + 模型名一致	`namespace` 必须带模型名，换模型后维度不同会炸

面试加分表述：知识库问答的缓存命中率往往比想象中高——用户问法集中在少数几十个高频问题上。我们上了 Redis LLM 缓存后，整体 token 成本降了约三成，P95 延迟因为命中直接返回也明显下降。但要注意缓存 key 必须包含「知识库版本号」，否则文档更新后会一直返回旧答案。

13. 配置动态化：一条链支持多模型

const configurableLlm = new ChatOpenAI({ temperature: 0 }).configurableFields({
  temperature: { id: "temperature", name: "采样温度" },
  modelName:   { id: "model",       name: "模型" },
})

const chain = prompt.pipe(configurableLlm).pipe(new StringOutputParser())

// 同一条链，运行时切换
await chain.invoke(input, { configurable: { model: "gpt-4o",      temperature: 0 } })
await chain.invoke(input, { configurable: { model: "gpt-4o-mini", temperature: 0.7 } })

配合 configurableAlternatives() 还能整体替换某个节点（比如 A/B 测试两版 Prompt），实现一次部署、配置驱动的灰度实验。

14. 并发与批量：离线评测的提速手段

// batch 自带并发控制
const results = await chain.batch(questions, { maxConcurrency: 5 })

// 容错批量：单条失败不中断整批
const results2 = await chain.batch(questions, { maxConcurrency: 5 }, { returnExceptions: true })
results2.forEach((r, i) => { if (r instanceof Error) logger.warn({ i, err: r }) })

场景	并发度建议
线上单请求内的扇出（多路检索）	不限，本来就只有几路
离线评测 / 批量灌库	按供应商 RPM 限额倒推，通常 3~10
Embedding 批量	用 `embedDocuments` 一次传多条，比并发多次请求更省且更快

⚠️ returnExceptions: true 是批量任务的必选项。灌库 1000 篇文档，第 300 篇解析失败就整批中断，会让任务永远跑不完。

15. 与 FastAPI / Node 服务对接：SSE 流式接口

// Node + Express：把 LCEL 的流接到 SSE
app.post("/chat", async (req, res) => {
  res.setHeader("Content-Type", "text/event-stream")
  res.setHeader("Cache-Control", "no-cache")
  res.setHeader("X-Accel-Buffering", "no")        // 关键：关掉 Nginx 缓冲，否则前端收不到增量
  res.flushHeaders()

  const ac = new AbortController()
  req.on("close", () => ac.abort())               // 用户关页面立刻中断，省 token

  try {
    for await (const ev of chain.streamEvents(req.body, { version: "v2", signal: ac.signal })) {
      if (ev.event === "on_retriever_end") res.write(`data: ${JSON.stringify({ type: "sources", sources: ev.data.output.map(toSource) })}\n\n`)
      if (ev.event === "on_chat_model_stream") res.write(`data: ${JSON.stringify({ type: "token", token: ev.data.chunk.content })}\n\n`)
    }
    res.write(`data: ${JSON.stringify({ type: "done" })}\n\n`)
  } catch (e) {
    res.write(`data: ${JSON.stringify({ type: "error", message: "生成失败，请重试" })}\n\n`)
  } finally {
    res.end()
  }
})

四个必踩的坑：

坑	解法
Nginx 缓冲导致流不出去	`X-Accel-Buffering: no` + `proxy_buffering off`
用户关闭页面但后端还在烧 token	监听 `req.on("close")` 传 `AbortSignal`
流中途报错，前端一直转圈	流内 try-catch，发一条 `type: error` 再 `end()`
网关超时切断长回答	定期发心跳注释行 `: ping\n\n`

16. 选型结论：框架薄、自研厚

Day25 已经给过结论，这里从框架视角再收一次口：

环节	用 LangChain	自研	理由
文档解析	❌	✅	官方 Loader 处理不了双栏/表格/扫描件
切片	⚠️ 起步可用	✅ 后期必改	业务文档的结构规则框架不懂
Embedding 调用	✅	❌	纯胶水，接口统一有价值
向量库读写	✅	❌	统一接口，换库成本低
检索 / 重排 / 融合	❌	✅	召回质量是 RAG 生死线，必须完全可控
Prompt 管理	✅	❌	模板 + partial + 占位符够用
编排 / 流式 / 重试	✅	❌	LCEL 是 LangChain 最值的部分
可观测	⚠️	✅	Callback 接口用框架的，上报落自家 APM
Agent	⚠️	—	固定流程不要上 Agent；要上优先 LangGraph

面试加分表述：我的判断是——LangChain 的编排层（LCEL）值得用，能力层（Loader/Splitter/预制 Chain）只适合做 PoC。所以我们的架构是「自研的解析和检索 + 实现框架接口 + 用 LCEL 串起来」，框架升级时受影响的只有薄薄一层适配代码，核心逻辑不跟着动。这比"全盘拥抱框架"和"完全不用框架"都更稳。

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：LangChain 是干什么的？核心组件有哪些？

精简口述答案（90 秒版）

LangChain 是大模型应用的标准接口层加一套编排 DSL。它本身不提供模型也不提供向量库，解决的是「胶水代码到处写、换供应商就得重构」的问题。核心组件我按数据流向记：Models 负责调模型，Prompts 负责模板化输入，DocumentLoaders 把各种格式读成统一的 Document，TextSplitters 做切片，Embeddings 转向量，VectorStores 存储和相似检索，Retrievers 是检索策略的统一抽象，Chains 也就是现在的 LCEL 负责编排，Memory 现在已经被 RunnableWithMessageHistory 取代，负责多轮上下文。贯穿全链路的数据结构是 Document，只有 pageContent 和 metadata 两个字段，metadata 承载来源、页码、租户这些工程化信息。

追问	答题要点
LCEL 是什么？	所有组件都实现 `Runnable` 接口，用 `pipe()`（Python 是 `\|`）串联；组合后自动继承 invoke/batch/stream/异步/重试/追踪
为什么不直接自己写？	自己写也能跑，但流式、重试、并发、追踪这些能力每条链都要重复实现一遍；而且换模型换库要改业务代码
那它有什么坏处？	封装黑盒，排查问题要读源码；版本迭代激进，API 频繁废弃；预制 Chain 的 Prompt 是英文的且不可控
新版包结构？	`@langchain/core` 放抽象零依赖，provider 独立成包，`@langchain/community` 放社区实现，`langchain` 放上层封装

✅ 项目关联话术：我们知识库项目最初直接调 OpenAI，后来因为合规要切国产模型。因为中间加了 LangChain 这层，模型和向量库的切换只改了工厂函数那一个文件，业务链路一行没动。反过来说，如果当初把 OpenAI SDK 调用散落在十几个 service 里，这次迁移至少要一周。

题 2：LCEL 和老的 Chain（LLMChain、ConversationChain）有什么区别？为什么要废弃老写法？

精简口述答案（90 秒版）

老 Chain 是「类继承式」的封装，每个场景一个类，输入输出都是字符串字典，key 靠约定，没有类型；想加流式、并行、重试都得框架自己给每个 Chain 实现一遍，没实现的就用不了。LCEL 换成了「组合式」：所有节点统一成 Runnable<I, O>，用 pipe 串起来。这样带来三个质变——第一，类型可推导，输入输出不再靠猜；第二，组合即继承能力，任意链路自动拥有 invoke、batch、stream、streamEvents 四种执行方式；第三，横切能力可以声明式挂载，withRetry、withFallbacks、withConfig 都是链上的方法。所以老 Chain 不是「不好用」，是「扩展性上限太低」，官方从 0.3 起全面标记废弃。

追问	答题要点
`RunnableParallel` 怎么用？	直接写对象字面量就会被自动转成并行节点，各 value 接收同一份输入、并发执行、结果按 key 合并
`RunnablePassthrough.assign()` 解决什么？	保留原始输入的同时追加计算字段，避免手写 `{ ...input, context }` 这种展开
为什么我的链不流式了？	中间有非流式节点（必须拿全量输入才能算的 Lambda），会把流堵住；要么透传，要么实现 `transform()`
`stream` 和 `streamEvents` 区别？	前者只给末端增量，后者给全链路生命周期事件，用来做"正在检索/正在生成"的进度提示

✅ 项目关联话术：我们接手时项目还是 LLMChain 写法，产品提了流式输出的需求，老 Chain 根本不支持，改造时顺手迁到了 LCEL。迁完之后一个意外收获是，前端的"正在检索知识库"进度提示直接用 streamEvents 的 on_retriever_end 事件就实现了，不用后端另外埋点推消息。

题 3：`RecursiveCharacterTextSplitter` 的分隔符顺序是怎么回事？中文为什么要特殊处理？

精简口述答案（90 秒版）

它的逻辑是按语义强度从强到弱递归降级。默认顺序是双换行、单换行、空格、空字符串：先用双换行切段落，切完每块如果还超过 chunkSize，就对这块用单换行切，还超就用空格切，最后兜底是逐字符硬切。这个顺序的含义是「尽量在语义边界切开」。问题在于默认分隔符是为英文设计的——中文正文里几乎没有空格，所以中文长段落走到第三级直接失效，掉进逐字符硬切，一句话被从中间劈开，这块的向量语义就毁了。解法是显式传入中文标点作为分隔符。另外 chunkSize 的单位是字符数不是 token 数，要精确控 token 得传自定义的 lengthFunction。

追问	答题要点
chunkOverlap 设多少？	经验值是 chunkSize 的 10%~20%；太小答案会被边界切断，太大召回重复内容浪费上下文
Markdown / 代码怎么切？	用 `MarkdownTextSplitter` 或 `fromLanguage("markdown"/"js")`，内置了结构级分隔符
切片大小怎么定？	小块检索准但上下文碎，大块反之；标准解法是 `ParentDocumentRetriever`——小块建索引、命中后返回父文档
切片质量怎么评？	抽样人工看「一个完整语义单元是否被切断」，再跑 Day25 的召回率评测对比不同参数

✅ 项目关联话术：我们一开始用默认配置，测试时发现"报销标准是多少"这个问题死活召回不到，排查后发现那段制度文字是一整个长段落，被按字符硬切成了三块，金额和条件分别落在不同块里。加上中文标点分隔符、再配合 ParentDocumentRetriever 之后，这类问题的召回准确率有明显提升。

题 4：Retriever 这个抽象的价值在哪？怎么自定义？

精简口述答案（90 秒版）

Retriever 是 LangChain 里设计最克制的抽象，整个接口只有一个方法：传进去一个 query 字符串，返回 Document 数组。正因为契约这么窄，任何检索逻辑都能伪装成 Retriever——向量检索、ES 关键词、图数据库、甚至调一个远程 HTTP 接口。它和 VectorStore 的关系是分层：VectorStore 管存储和最基础的相似搜索，Retriever 管检索策略，asRetriever() 把前者包成后者。自定义只需要继承 BaseRetriever 实现 _getRelevantDocuments，注意是带下划线的内部方法，因为基类在外层包了 Callback 和链路追踪，直接重写 invoke 会让追踪断掉。

追问	答题要点
官方高级 Retriever 有哪些？	MultiQuery（问题改写多路召回）、ParentDocument（小块索引大块返回）、SelfQuery（自然语言转结构化过滤）、ContextualCompression（召回后裁剪）、Ensemble（多路 RRF 融合）
你会用官方的还是自研？	自研。混合检索、重排、权限过滤是召回质量的生死线，必须完全可控；但包装成 Retriever 接口接入 LCEL
打分阈值放哪一层？	放 Retriever 里，把 score 写进 `metadata.score`，上游凭此决定是否拒答
权限过滤怎么做？	向量库的 `filter` 参数带 `tenantId`/`deptId`，必须在检索时过滤而不是检索后过滤，否则 topK 会被无权限文档占满

✅ 项目关联话术：我们的 HybridRetriever 内部是向量 20 路加 ES 20 路、RRF 融合、重排模型取 top6、再按 0.35 分数阈值过滤。这一整套是自研的，但对外只暴露 Retriever 接口。好处是框架升级时它完全不受影响，同时又能免费拿到 LCEL 的流式和 LangSmith 的链路追踪。

题 5：LangChain 的 Memory 怎么实现的？多轮对话 token 爆炸怎么办？

精简口述答案（90 秒版）

要先说清现状：旧的 BufferMemory 加 ConversationChain 已经废弃了，现在的标准做法是 RunnableWithMessageHistory——它包在链外面，根据 config 里传的 sessionId 去取历史、把历史注入到 MessagesPlaceholder 占位符，调用结束后自动把本轮的问和答追加回历史。token 爆炸的本质是全量缓冲下历史长度线性增长，轮数一多必然超上下文窗口，而且每轮都要为全部历史付费。三种策略：滑动窗口只留最近 N 轮，成本恒定但丢早期信息；摘要压缩把旧历史用 LLM 压成一段摘要，成本可控但摘要会失真；最实用的是组合——最近三轮保留原文，更早的滚动摘要，外面再加一个硬性 token 上限做截断兜底。

追问	答题要点
sessionId 为什么从 config 传不从 input 传？	它是运行时上下文而非业务入参；放 config 里才能被 Callback、追踪统一透传，也不污染 Prompt 变量
多实例部署怎么办？	`InMemoryChatMessageHistory` 只能单进程用，必须换 `RedisChatMessageHistory`，否则负载均衡打到别的实例就失忆
历史和 RAG 怎么配合？	两件事：一是用历史做 Query 改写（`createHistoryAwareRetriever`）解决指代；二是把历史作为消息注入 Prompt。不要把历史也拿去向量检索
截断时要注意什么？	不能从中间切断一对问答，必须成对丢弃；SystemMessage 永远保留

✅ 项目关联话术：我们线上出过一次事故，客服会话平均二十多轮，用的是全量缓冲，结果上下文超限直接 400，而且账单比预估高了一倍多。改成「最近 3 轮原文 + 滚动摘要 + 6k token 硬上限」之后，单会话成本稳定了，长会话也不再报错。摘要本身用的是便宜的小模型，只在历史超过阈值时才触发。

题 6：多轮对话里用户说"它支持哪些格式"，你怎么保证检索准确？

精简口述答案（90 秒版）

这是 Query 改写要解决的问题。直接拿"它支持哪些格式"去做向量检索，query 里没有任何实体词，召回必然是噪声。标准做法是在检索前插一步：把对话历史和当前问题一起给 LLM，让它改写成一个不依赖上下文的独立问题，比如"文档解析服务支持哪些文件格式"，再拿改写后的 query 去检索。LangChain 里有现成的 createHistoryAwareRetriever，它内部的逻辑就是：如果没有历史就直接走原 Retriever，有历史就先过一遍改写 Prompt。改写这一步要用 temperature 0 的小模型，并且要约束它只输出问题本身，不要解释。

追问	答题要点
改写会不会引入新问题？	会。模型可能改写过度，把用户没问的信息塞进去。缓解：Prompt 里强调"只补全指代，不要扩展语义"，并保留原 query 一起做多路召回
多了一次 LLM 调用，延迟怎么办？	用小模型（成本和延迟都是主模型的几分之一）；首轮无历史时跳过；改写和"原 query 检索"可以并行
还有别的改写方式吗？	多查询扩展（一个问题改成 N 个不同角度）、HyDE（先让模型生成假设答案再用答案去检索）、关键词抽取补充 BM25
怎么验证改写有效？	构造多轮测试集，对比"改写前/改写后"的召回命中率，这是 Day25 三层评测里检索层的指标

✅ 项目关联话术：我们做过对比，多轮场景下不做改写的召回命中率明显偏低，几乎所有带指代的问题都答不对。上了改写之后这类问题基本能正常回答。我们还做了个优化：把改写后的 query 和原始 query 同时送进混合检索做 RRF 融合，避免改写失真时直接丢掉正确答案。

题 7：LangChain 的链路怎么做可观测？出了问题怎么排查？

精简口述答案（90 秒版）

两个层面。框架层靠 Callback，它是所有 Runnable 的统一钩子，能拿到 LLM 开始/每个 token/结束/报错、Chain 开始结束、Tool 调用、Retriever 召回这些事件，每个事件带 runId 和 parentRunId，天然构成一棵调用树。我们在 handleLLMEnd 里统计 token 和耗时，在 handleRetrieverEnd 里统计召回条数，在 handleChainError 里打错误日志，再通过 metadata 把业务 traceId 透传进来和 APM 打通。产品层靠 LangSmith，只要配三个环境变量就自动上报，能看到每个节点的输入输出全文、耗时火焰图、token 费用，失败的 case 可以一键复现，还能把线上 case 攒成数据集跑回归评测。但 LangSmith 是 SaaS，企业数据合规通常过不了，实际落地更多是用自定义 Callback 上报到自家 ES，或者走 OpenTelemetry 的 GenAI 语义约定接到现有 APM。

追问	答题要点
Callback 挂载的两种方式？	`withConfig({ callbacks })` 构造期全局生效；`invoke(input, { callbacks })` 调用期单次生效
Callback 会影响性能吗？	会，默认同步阻塞执行。里面不能写数据库或做慢 IO，必须投递到队列异步处理
没有 LangSmith 怎么调试？	`returnIntermediateSteps`（Agent）、`streamLog()` 看增量状态、在关键节点插 `RunnableLambda` 打日志
RAG 答错了怎么定位是检索问题还是生成问题？	看追踪里 Retriever 的输出：召回里没有正确文档就是检索问题，有但答错就是生成/Prompt 问题。这是 Day25 分层评测的落地抓手

✅ 项目关联话术：我们早期排查"答错"全靠人工复现，效率极低。后来写了一个自定义 Callback，把每次请求的 runId、改写后的 query、召回的 chunk id 列表、最终答案、token 消耗写进 ES，前端展示时把 runId 带上。这样用户一反馈"这个答案不对"，我们直接按 runId 查出当时召回了什么，十几秒就能判断是检索召不回还是模型没按上下文答。

题 8：什么时候该用 Agent，什么时候不该用？

精简口述答案（90 秒版）

Agent 的本质是一个 while 循环：把工具列表的 name、description、schema 一起给模型，模型返回 tool_calls 就执行工具、把结果作为 ToolMessage 追加回消息列表再问一次，直到模型不再要求调工具。它的适用前提是「执行路径无法事先确定」。但代价很大：不可控，同一个问题两次可能走不同路径；贵，每轮循环都是一次完整 LLM 调用；难调试，中间步骤是黑盒；还有安全问题，退款删除这类高危操作会被模型直接执行。所以我的判断是：流程固定的场景（比如纯知识库问答）绝对不要上 Agent，用 LCEL 写死链路，又快又便宜又可测；需要动态决策但步骤可枚举的，用 LangGraph，把隐式循环变成显式状态图，支持分支、检查点、人工介入；只有完全开放的探索型任务才用 AgentExecutor，而且生产环境要慎用。

追问	答题要点
工具描述怎么写？	`description` 就是 Agent 的源码，必须写清「什么时候该调、什么时候不该调」；参数用 zod 的 `.describe()` 说明格式
怎么防止无限循环？	`maxIterations` 必须设（一般 5 以内）；工具数量控制在 10 个以内，太多模型选不准
高危操作怎么办？	工具分级：只读工具直接执行，写操作工具返回"待确认"结构体，由前端弹确认框，用户点了才真正执行
LangGraph 比 AgentExecutor 强在哪？	显式状态图 + 检查点持久化，可以中断后恢复，可以插入 human-in-the-loop 节点，状态可观测

✅ 项目关联话术：产品一开始要求做成"全能助手"，我们用 Agent 做了版本，上线前压测发现平均 3.2 轮循环、延迟超过 8 秒、成本是纯 RAG 的三倍多，而且偶发性乱调工具。最后改成用小模型做意图分类，再用 RunnableBranch 路由到订单链、知识库链、闲聊链三条固定链路，延迟和成本都降回来了，准确率反而更高。只有"跨多个系统的复杂查询"这一类才走 Agent。

题 9：LangChain 这层封装有什么代价？你们怎么权衡的？

精简口述答案（90 秒版）

四个代价。第一是黑盒：预制 Chain 里的 Prompt 是英文硬编码的，中文场景效果打折且不好改，排查问题经常要去读框架源码。第二是版本激进：0.0 到 0.3 之间大量 API 废弃、包结构重拆，升级成本不低，而且 @langchain/core 版本不一致会出现两份 Runnable 基类，组合时直接报错。第三是性能开销：每个节点都包了 Callback 和追踪，虽然单次开销不大但链路长了会累积。第四是「能力层」质量有限，官方的 Loader 处理不了双栏 PDF 和表格，Splitter 不懂业务文档结构。我们的权衡是框架薄、自研厚：编排层用 LCEL，这是 LangChain 最值的部分；能力层里纯胶水的（Embedding 调用、向量库读写、Prompt 模板）用框架；解析和检索这两个决定效果的环节完全自研，然后实现框架接口接进去。这样框架升级只影响薄薄一层适配代码。

追问	答题要点
版本怎么管？	锁定精确版本不用 `^`；`@langchain/core` 在自研包里声明为 peerDependency；升级前先在评测集上跑回归
不用框架可以吗？	可以，直接调 SDK 完全可行。但流式、重试、并发、追踪要自己实现一遍；小项目自研更清爽，多链路多模型时框架更划算
为什么不用更轻的方案？	可以只用 `@langchain/core` 的 Runnable 做编排，不装 community，这就是最薄的用法
预制 Chain 的 Prompt 怎么改？	`createStuffDocumentsChain` 等都支持传入自己的 `prompt`，一定要传中文自定义版，不要用默认的

✅ 项目关联话术：我们明确约定了一条边界——任何影响召回质量和答案忠实度的逻辑都不许交给框架。所以切片规则、混合检索、重排、拒答阈值、引用溯源全是自研并且有单测覆盖的；框架只负责把这些模块串成链、提供流式和追踪。实际升级过两次大版本，核心逻辑都没动过，只改了几个 import 路径和一个 Retriever 基类的方法签名。

题 10：怎么给 LangChain 应用做成本控制？

精简口述答案（90 秒版）

分三层。第一层是减少调用次数：最有效的是 LLM 结果缓存，把 Prompt 做 key 存 Redis，知识库问答的用户问法高度集中在少数高频问题上，命中率比想象的高；Embedding 也要缓存，同一段文本重复灌库没必要重复付费，用 CacheBackedEmbeddings 配字节存储，namespace 必须带模型名。第二层是减少单次 token：上下文组装时按相关度截断而不是全塞，历史用摘要压缩，Prompt 里的格式说明能用 withStructuredOutput 走原生能力就不要在 Prompt 里写一长段 schema 描述。第三层是模型分级：意图分类、Query 改写、摘要这些辅助环节用便宜的小模型，只有最终生成用大模型，再配 withFallbacks 在大模型限流时降级。统计口径靠 Callback 累加 llmOutput.tokenUsage，按 tenantId 打标签，才能算出单会话成本和各环节占比。

追问	答题要点
LLM 缓存的坑？	Prompt 里带时间戳就永远不命中；缓存 key 必须包含知识库版本号，否则文档更新后一直返回旧答案；temperature>0 时缓存会抹掉多样性
语义缓存靠谱吗？	阈值高了没效果、低了答非所问，必须人工校验高频问题白名单，不要全量开
流式场景怎么省？	监听连接关闭传 `AbortSignal` 立刻中断生成，用户关页面后继续生成是纯浪费
怎么知道钱花在哪？	Callback 里按 `tags` 和 `metadata` 分维度上报，区分改写/生成/摘要各环节的 token 占比

✅ 项目关联话术：我们上了 Redis LLM 缓存加 Embedding 缓存之后，整体 token 成本降了约三成；再把 Query 改写和会话摘要从主模型换成小模型，又降了一部分。最关键的一步是先把 Callback 的分维度统计做出来——在那之前我们根本不知道改写环节占了多少成本，优化全靠猜。

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：用 LCEL 写一条最小可用的 RAG 链

要求：输入 { question }，输出带引用标号的答案字符串；检索为空时直接拒答，不调用 LLM。

import { ChatPromptTemplate } from "@langchain/core/prompts"
import { StringOutputParser } from "@langchain/core/output_parsers"
import { RunnableSequence, RunnableLambda } from "@langchain/core/runnables"
import type { Document } from "@langchain/core/documents"

const REFUSE = "抱歉，知识库中没有找到相关依据，无法回答该问题。"

const prompt = ChatPromptTemplate.fromMessages([
  ["system", [
    "你是企业知识库助手。严格遵守：",
    "1. 只能依据 <context> 中的内容回答，不得使用任何外部知识；",
    "2. 每个结论后用 [编号] 标注依据的片段，编号对应 context 中的序号；",
    "3. 若 context 不足以回答，只输出一句：" + REFUSE,
  ].join("\n")],
  ["human", "<context>\n{context}\n</context>\n\n问题：{question}"],
])

// 把 Document[] 渲染成带编号的上下文
const formatDocs = (docs: Document[]) =>
  docs.map((d, i) => `[${i + 1}] 来源：${d.metadata.source ?? "未知"}\n${d.pageContent}`).join("\n\n")

const answerChain = prompt.pipe(llm).pipe(new StringOutputParser())

export const ragChain = RunnableSequence.from([
  // step1：检索，结果和原问题一起往下传
  RunnableLambda.from(async (input: { question: string }) => ({
    question: input.question,
    docs: await retriever.invoke(input.question),
  })),
  // step2：空召回直接短路，不浪费一次 LLM 调用
  RunnableLambda.from(async ({ question, docs }) => {
    if (!docs.length) return REFUSE
    return answerChain.invoke({ question, context: formatDocs(docs) })
  }),
])

// 调用
const answer = await ragChain.invoke({ question: "差旅住宿标准是多少？" })

更紧凑的等价写法（无短路需求时）：

const ragChain = RunnablePassthrough
  .assign({ context: async (i) => formatDocs(await retriever.invoke(i.question)) })
  .pipe(prompt).pipe(llm).pipe(new StringOutputParser())

边界 Case 自测清单

[ ] 检索结果为空 → 直接返回拒答文案，且没有产生 LLM 调用（用 Callback 计数验证）
[ ] 问题里包含 { } → 不会被模板引擎当插值解析（变量值不参与二次插值，但模板常量里的花括号要写 {{}}）
[ ] 召回文档超长 → 组装前按 token 预算截断，而不是等模型报上下文超限
[ ] metadata.source 缺失 → 不抛错，降级成"未知"
[ ] 答案里出现了 context 中不存在的编号 → 后置校验，过滤掉非法引用标号
[ ] LLM 超时 → 链外层有 withRetry + withFallbacks，且最终失败返回友好文案而非堆栈

变种拓展题

改成流式输出，并且在正文开始前先把召回来源推给前端（用 streamEvents 的 on_retriever_end）。
用 withStructuredOutput 把输出改成 { answer, citations: number[], refused: boolean }，并校验 citations 都在合法范围内。
加入多轮：用 RunnableWithMessageHistory 包一层，并用 createHistoryAwareRetriever 做 Query 改写。
把 RunnableLambda 的短路逻辑改用 RunnableBranch 实现，对比可读性。

题 2：实现一个混合检索的自定义 Retriever

要求：向量检索 + 关键词检索并发，RRF融合，重排取 topK，低分过滤，全程不破坏链路追踪。

import { BaseRetriever, type BaseRetrieverInput } from "@langchain/core/retrievers"
import type { Document } from "@langchain/core/documents"
import type { CallbackManagerForRetrieverRun } from "@langchain/core/callbacks/manager"

interface Deps {
  vectorStore: { similaritySearch(q: string, k: number, filter?: object): Promise<Document[]> }
  keywordSearch(q: string, k: number, filter?: object): Promise<Document[]>
  rerank?(q: string, docs: Document[], k: number): Promise<Array<Document & { score: number }>>
}

interface Options extends BaseRetrieverInput {
  recallK?: number      // 每路召回量
  topK?: number         // 最终返回量
  minScore?: number     // 分数阈值，低于则丢弃（为拒答留出口）
  filter?: object       // 权限 / 状态过滤，必须下推到检索层
}

export class HybridRetriever extends BaseRetriever {
  lc_namespace = ["app", "retrievers", "hybrid"]
  private opt: Required<Pick<Options, "recallK" | "topK" | "minScore">> & { filter?: object }

  constructor(private deps: Deps, options: Options = {}) {
    super(options)
    this.opt = {
      recallK: options.recallK ?? 20,
      topK: options.topK ?? 6,
      minScore: options.minScore ?? 0.35,
      filter: options.filter,
    }
  }

  async _getRelevantDocuments(query: string, runManager?: CallbackManagerForRetrieverRun) {
    const { recallK, topK, minScore, filter } = this.opt

    // 1. 两路并发召回，单路失败不拖垮整体
    const [vecRes, kwRes] = await Promise.allSettled([
      this.deps.vectorStore.similaritySearch(query, recallK, filter),
      this.deps.keywordSearch(query, recallK, filter),
    ])
    const pick = (r: PromiseSettledResult<Document[]>) => (r.status === "fulfilled" ? r.value : [])
    const lists = [pick(vecRes), pick(kwRes)].filter(l => l.length)
    if (!lists.length) return []

    // 2. RRF 融合：score = Σ 1 / (k + rank)
    const fused = this.rrf(lists, 60)

    // 3. 重排（可选），没有重排模型就用融合分
    const scored = this.deps.rerank
      ? await this.deps.rerank(query, fused.slice(0, recallK), topK)
      : fused.slice(0, topK).map((d, i) => Object.assign(d, { score: 1 - i / topK }))

    // 4. 阈值过滤 + 把分数写回 metadata，供上游拒答判断
    return scored
      .filter(d => d.score >= minScore)
      .map(d => ({ ...d, metadata: { ...d.metadata, score: d.score } }))
  }

  private rrf(lists: Document[][], k = 60): Document[] {
    const table = new Map<string, { doc: Document; s: number }>()
    for (const list of lists) {
      list.forEach((doc, rank) => {
        const id = this.idOf(doc)
        const cur = table.get(id) ?? { doc, s: 0 }
        cur.s += 1 / (k + rank + 1)
        table.set(id, cur)
      })
    }
    return [...table.values()].sort((a, b) => b.s - a.s).map(x => x.doc)
  }

  // 去重键：优先业务 chunkId，退化到内容哈希，避免同一片段被两路重复计入
  private idOf(doc: Document): string {
    return doc.metadata?.chunkId ?? `${doc.metadata?.source ?? ""}#${doc.pageContent.slice(0, 64)}`
  }
}

边界 Case 自测清单

[ ] 向量库挂了但 ES 正常 → 仍返回关键词结果（allSettled 而非 all）
[ ] 两路都挂 → 返回空数组而不是抛错，上游走拒答
[ ] 同一片段被两路召回 → idOf 正确去重，RRF 分数累加而不是出现重复条目
[ ] filter 为空 → 不会把过滤条件写成 undefined 传给向量库导致全表扫
[ ] 所有结果分数都低于 minScore → 返回空数组（这是拒答的关键入口，不要为了"有结果"去掉阈值）
[ ] 重排模型超时 → 降级使用融合分，不要让整次检索失败
[ ] 通过 invoke() 调用时 → LangSmith 里能看到独立的 retriever 节点（证明没有重写 invoke）

变种拓展题

加入权重：向量和关键词按 0.7 / 0.3 加权，而不是等权 RRF，对比两者在评测集上的差异。
支持多 query：接收改写后的 N 个 query 并发检索，再统一融合。
加一层缓存：相同 query + filter 在 60 秒内直接返回上次结果。
实现 ParentDocumentRetriever 的效果——命中小块后去 docstore 取父文档并按 docId 去重。

题 3：实现带摘要压缩与 token 上限的会话历史

要求：最近 N 轮保留原文，更早历史滚动摘要，整体不超过 token 预算；可接入 RunnableWithMessageHistory。

import { BaseListChatMessageHistory } from "@langchain/core/chat_history"
import { AIMessage, HumanMessage, SystemMessage, type BaseMessage } from "@langchain/core/messages"

interface Cfg {
  keepRounds: number      // 保留原文的轮数（1 轮 = 1 问 + 1 答）
  maxTokens: number       // 历史总预算
  summarize(text: string): Promise<string>   // 用小模型做摘要
  estimate(text: string): number             // token 估算，中文约 1 字 ≈ 1 token
}

export class CompressedHistory extends BaseListChatMessageHistory {
  lc_namespace = ["app", "history"]
  private messages: BaseMessage[] = []
  private summary = ""

  constructor(public sessionId: string, private cfg: Cfg) { super() }

  async getMessages(): Promise<BaseMessage[]> {
    const head = this.summary ? [new SystemMessage(`以下是此前对话的摘要：\n${this.summary}`)] : []
    return [...head, ...this.messages]
  }

  async addMessage(message: BaseMessage) {
    this.messages.push(message)
    await this.compress()
  }

  private async compress() {
    const keep = this.cfg.keepRounds * 2

    // 1. 超出保留轮数的部分，滚动并入摘要
    if (this.messages.length > keep) {
      const evicted = this.messages.splice(0, this.messages.length - keep)
      const text = evicted.map(m => `${m._getType()}: ${m.content}`).join("\n")
      this.summary = await this.cfg.summarize(
        this.summary ? `已有摘要：\n${this.summary}\n\n新增对话：\n${text}` : text,
      )
    }

    // 2. 仍超预算时，成对丢弃最旧的问答（绝不从中间切断一对）
    while (this.total() > this.cfg.maxTokens && this.messages.length >= 2) {
      this.messages.splice(0, 2)
    }

    // 3. 摘要自身也可能膨胀，兜底二次压缩
    if (this.cfg.estimate(this.summary) > this.cfg.maxTokens / 2) {
      this.summary = await this.cfg.summarize(`请把下面摘要压缩到原来的一半：\n${this.summary}`)
    }
  }

  private total() {
    return this.cfg.estimate(this.summary) +
      this.messages.reduce((s, m) => s + this.cfg.estimate(String(m.content)), 0)
  }

  async clear() { this.messages = []; this.summary = "" }
}

接入链路：

const stores = new Map<string, CompressedHistory>()
const chat = new RunnableWithMessageHistory({
  runnable: ragChain,
  getMessageHistory: async (sessionId) => {
    if (!stores.has(sessionId)) stores.set(sessionId, new CompressedHistory(sessionId, cfg))
    return stores.get(sessionId)!
  },
  inputMessagesKey: "question",
  historyMessagesKey: "history",
})
await chat.invoke({ question: "它的上限呢？" }, { configurable: { sessionId: "u_1001" } })

边界 Case 自测清单

[ ] 首轮对话 → 不触发摘要，不产生额外 LLM 调用
[ ] 恰好等于 keepRounds → 不触发压缩（边界用 > 而非 >=）
[ ] 单条消息本身就超预算（用户粘贴万字文本） → 不能死循环，需对单条做截断
[ ] 丢弃时 → 永远成对丢弃，不出现"有问无答"的残缺历史
[ ] 摘要调用失败 → 降级为直接丢弃旧消息，不能让整次对话失败
[ ] 多实例部署 → 换成 Redis 实现后，同一 sessionId 在不同实例读到同一份历史
[ ] 并发同一 sessionId 两个请求 → 需加锁或用乐观版本号，否则历史会互相覆盖

变种拓展题

换成 Redis 存储：摘要和消息分别用 String 和 List，加 TTL 自动过期。
把 token 估算换成真实 tokenizer（js-tiktoken），对比估算误差。
只在「历史超过阈值」时才摘要，并把摘要操作放进队列异步执行，不阻塞当前回答。
支持"重置上下文"指令：用户说"换个话题"时清空历史但保留摘要作为长期记忆。

题 4：给链路加上重试、降级、超时、可观测四件套

要求：主模型失败自动降级，超时可中断，全链路埋点统计 token 与耗时。

import { BaseCallbackHandler } from "@langchain/core/callbacks/base"
import type { LLMResult } from "@langchain/core/outputs"

class MetricsHandler extends BaseCallbackHandler {
  name = "metrics"
  private t = new Map<string, number>()
  constructor(private ctx: { traceId: string; tenantId: string }) { super() }

  handleLLMStart(_l: any, _p: string[], runId: string) { this.t.set(runId, Date.now()) }

  handleLLMEnd(out: LLMResult, runId: string) {
    const u = out.llmOutput?.tokenUsage
    // 注意：只做内存累加或投队列，绝不在这里写库
    queue.push({
      ...this.ctx, runId, kind: "llm",
      cost: Date.now() - (this.t.get(runId) ?? Date.now()),
      prompt: u?.promptTokens ?? 0, completion: u?.completionTokens ?? 0,
    })
    this.t.delete(runId)
  }

  handleRetrieverEnd(docs: any[], runId: string) {
    queue.push({ ...this.ctx, runId, kind: "retriever", hits: docs.length,
      scores: docs.map(d => d.metadata?.score) })
  }

  handleChainError(err: Error, runId: string) {
    logger.error({ ...this.ctx, runId, msg: err.message }, "chain failed")
  }
}

// 模型侧：关掉内置重试，避免与 withRetry 相乘
const primary = new ChatOpenAI({ modelName: "gpt-4o", temperature: 0, timeout: 20_000, maxRetries: 0 })
const backup  = new ChatOpenAI({ modelName: "gpt-4o-mini", temperature: 0, timeout: 20_000, maxRetries: 0 })

const guardedLlm = primary
  .withRetry({ stopAfterAttempt: 2 })   // 瞬时错误重试
  .withFallbacks([backup])              // 仍失败则降级

export const robustChain = prompt.pipe(guardedLlm).pipe(new StringOutputParser())

// 调用：传入可观测上下文 + 中断信号
export async function ask(question: string, ctx: { traceId: string; tenantId: string }, signal: AbortSignal) {
  return robustChain.invoke({ question }, {
    signal,
    callbacks: [new MetricsHandler(ctx)],
    metadata: { ...ctx },
    tags: ["rag", "v2"],
    runName: "kb-qa",
  })
}

边界 Case 自测清单

[ ] 主模型 429 → 重试 2 次后降级到备用模型，最终有结果
[ ] 主模型 400（参数错误） → 不应重试，直接抛出（必要时自定义 shouldRetry 判断）
[ ] 真实请求次数 → maxRetries: 0 已关闭 SDK 内置重试，总次数等于 stopAfterAttempt，不是相乘
[ ] 客户端断开 → AbortSignal 触发后生成立即停止，不继续产生 token 费用
[ ] Callback 内部抛错 → 不能让主链路失败，需在 handler 内自行 try-catch
[ ] 降级发生时 → 日志/埋点里能区分出是主模型还是备用模型产生的答案
[ ] 高并发 → MetricsHandler 的 Map 按 runId 隔离，且结束时 delete 防内存泄漏

变种拓展题

加「上下文超长」专项降级：捕获长度错误后自动截断上下文重试一次。
用 configurableFields 让模型与温度从配置中心下发，实现不重启灰度。
把埋点改成 OpenTelemetry Span，父子关系用 runId / parentRunId 映射。
给整条链加熔断：错误率超阈值时直接返回兜底文案，不再打模型。

题 5：用 `streamEvents` 实现「先出来源、再流正文」的 SSE 接口

要求：前端依次收到来源列表 → 正文 token 流 → 实际引用 → 结束；中途报错有兜底，用户关闭连接立即中断。

type SsePayload =
  | { type: "sources"; sources: Array<{ idx: number; title: string; url?: string; score?: number }> }
  | { type: "token"; token: string }
  | { type: "citations"; used: number[] }
  | { type: "error"; message: string }
  | { type: "done" }

const send = (res: any, p: SsePayload) => res.write(`data: ${JSON.stringify(p)}\n\n`)

app.post("/api/chat", async (req, res) => {
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    Connection: "keep-alive",
    "X-Accel-Buffering": "no",        // 关键：禁用 Nginx 缓冲
  })

  const ac = new AbortController()
  req.on("close", () => ac.abort())   // 用户关页面 → 立即停止生成
  const heartbeat = setInterval(() => res.write(": ping\n\n"), 15_000)  // 防网关超时

  let full = ""
  try {
    const stream = ragChain.streamEvents(
      { question: req.body.question, history: req.body.history ?? [] },
      { version: "v2", signal: ac.signal, metadata: { traceId: req.id } },
    )

    for await (const ev of stream) {
      switch (ev.event) {
        case "on_retriever_end":
          send(res, {
            type: "sources",
            sources: (ev.data.output as any[]).map((d, i) => ({
              idx: i + 1,
              title: d.metadata?.title ?? d.metadata?.source ?? "未知来源",
              url: d.metadata?.url,
              score: d.metadata?.score,
            })),
          })
          break
        case "on_chat_model_stream": {
          const token = String(ev.data.chunk?.content ?? "")
          if (!token) break
          full += token
          send(res, { type: "token", token })
          break
        }
      }
    }

    // 正文流完，回填"实际被引用"的编号，前端据此高亮
    const used = [...new Set([...full.matchAll(/\[(\d+)\]/g)].map(m => Number(m[1])))].sort((a, b) => a - b)
    send(res, { type: "citations", used })
    send(res, { type: "done" })
  } catch (e: any) {
    if (e?.name === "AbortError") return               // 用户主动断开，不算错误
    logger.error({ traceId: req.id, err: e }, "sse failed")
    send(res, { type: "error", message: "生成失败，请稍后重试" })
  } finally {
    clearInterval(heartbeat)
    res.end()
  }
})

边界 Case 自测清单

[ ] 经过 Nginx → 前端能收到增量而非一次性返回（X-Accel-Buffering: no + proxy_buffering off）
[ ] 用户刷新页面 → 后端生成在 1 秒内停止，日志里能看到中断记录
[ ] 流到一半模型报错 → 前端收到 type: error 而不是永远转圈
[ ] 长回答超过网关超时 → 心跳注释行保持连接存活
[ ] 空召回 → 先推空 sources，再流拒答文案，前端不展示引用区
[ ] 答案里引用了不存在的编号 → citations 过滤掉越界编号后再下发
[ ] token 内容包含 \n 或 data: 字样 → 已被 JSON.stringify 转义，不会破坏 SSE 协议
[ ] 同一连接重复 res.end() → finally 只执行一次，无 "write after end" 报错

变种拓展题

换成 Web Streams（ReadableStream）实现，适配 Next.js Route Handler / Edge Runtime。
加断线重连：用 SSE 的 Last-Event-ID 从上次断点续传已生成的内容。
把 on_tool_start / on_tool_end 也推给前端，展示 Agent 的工具调用过程。
在服务端做敏感词流式拦截：命中时立即中断并替换为合规话术。

⚠️ 当日踩坑实录

版本与依赖

❌ 坑	✅ 解决方案
照着教程 `import { ChatOpenAI } from "langchain/chat_models/openai"`，报找不到模块	0.1 之后包结构重拆，改成 `@langchain/openai`；核心抽象在 `@langchain/core`
`pipe()` 时报 "Expected a Runnable"，但明明传的是 Runnable	依赖树里装了两份 `@langchain/core`，基类不是同一个。用 `npm ls @langchain/core` 排查，锁成单一版本（pnpm 用 overrides）
自研工具包发布后，使用方出现同样的双实例问题	自研包把 `@langchain/core` 声明为 `peerDependencies` 而不是 `dependencies`
升级小版本后一堆方法报废弃警告甚至直接报错	LangChain 迭代激进，依赖必须锁精确版本不用 `^`；升级前先在评测集上跑回归
官方示例是 Python 的，照抄到 JS 不能用	两端 API 不完全对齐：Python 用 `\|` 和 `__or__`，JS 用 `.pipe()`；Python 的 `get_openai_callback` 在 JS 里没有，要自己写 Callback 累加

Prompt 与模板

❌ 坑	✅ 解决方案
Prompt 里写了 JSON 示例，报 "Missing value for input variable"	`{}` 是插值语法，字面花括号必须写成 `{{` `}}` 转义
用 `fromTemplate` 拼多轮对话，历史被当成普通字符串转义，模型分不清角色	改用 `ChatPromptTemplate.fromMessages` + `MessagesPlaceholder`，它接收 `BaseMessage[]`
中途插入第二条 `SystemMessage` 想改变角色，模型完全无视	多数模型只认第一条 system；要中途加约束就写进 human 消息里
用了 `createStuffDocumentsChain` 默认 Prompt，中文回答夹英文、不肯拒答	预制 Chain 的默认 Prompt 是英文硬编码的，必须传自己的中文 prompt
用户输入里带 `{{` 导致模板解析异常	插值的变量值不会被二次解析，但要确认没有把用户输入拼进模板字符串本身

切片与检索

❌ 坑	✅ 解决方案
中文文档被切得稀碎，一句话从中间断开	默认分隔符 `["\n\n","\n"," ",""]` 为英文设计，中文无空格直接掉到逐字符硬切；显式加 `。！？；，`
`chunkSize: 500` 以为是 500 token，结果上下文反复超限	单位是字符数；要精确控 token 得传 `lengthFunction` 用 tokenizer 计算
`chunkOverlap` 设成 chunkSize 的一半，召回全是重复内容	经验值 10%~20%
Markdown 的标题和正文被切开，chunk 丢失了归属上下文	用 `MarkdownTextSplitter`，或切片时把标题路径写进 `metadata` 并在组装上下文时拼回正文前面
切片越小召回越准但答案不完整	用 `ParentDocumentRetriever`：小块建索引、命中后返回父文档
权限过滤写在检索之后	topK 会被无权限文档占满，导致有权限的内容根本没进候选。`filter` 必须下推到向量库检索时
换了 embedding 模型，检索结果全乱	维度和语义空间都变了，必须全量重新灌库；Embedding 缓存的 namespace 要带模型名
官方 `PDFLoader` 读双栏 PDF，文字左右交错成乱序	Loader 只解决"读得出"，不解决"读得对"；复杂版式必须自研解析后手动 `new Document()`

LCEL 与链路

❌ 坑	✅ 解决方案
还在用 `LLMChain` / `ConversationChain`，想加流式发现不支持	旧 Chain 已废弃，迁到 LCEL；`chain.call({})` 返回 `{ text }` 的猜 key 时代结束了
链路明明用了 `stream()`，前端还是一次性拿到全部内容	中间有非流式节点（需要全量输入的 Lambda）把流堵住了；改成透传或实现 `transform()`
想展示"正在检索知识库"，不知道怎么拿中间状态	用 `streamEvents({ version: "v2" })` 监听 `on_retriever_end` 等事件
对象字面量里写了异步函数，不确定是串行还是并行	对象字面量会被转成 `RunnableParallel`，各 value 并发执行
手写 `{ ...input, context }` 展开到处都是	用 `RunnablePassthrough.assign({ context })`，保留原输入并追加字段
`withRetry` 配了 3 次，日志里看到 9 次请求	模型自身 `maxRetries` 默认不为 0，两层重试相乘；配 `withRetry` 时把 `maxRetries: 0`
对 400 参数错误也在重试，白烧配额	重试要区分错误类型：429/5xx/网络抖动才重试，400/审核拒绝直接抛
自定义 Retriever 重写了 `invoke`，LangSmith 里追踪断了	必须实现 `_getRelevantDocuments`，基类在外层包了 Callback 与追踪
一条链同时支持两个模型，只能复制两份	`configurableFields()` + `invoke(input, { configurable })` 运行时切换

Memory 与多轮

❌ 坑	✅ 解决方案
`BufferMemory` 无限追加，二十轮后上下文超限、账单翻倍	全量缓冲 token 线性增长。换成「最近 N 轮原文 + 滚动摘要 + 硬性 token 上限」三重策略
截断历史时从中间切断，出现"有问无答"的残缺对话	必须成对丢弃问答，SystemMessage 永远保留
多实例部署后用户反馈"机器人失忆"	`InMemoryChatMessageHistory` 只在单进程有效，换 `RedisChatMessageHistory`
`sessionId` 当成普通入参放进 `invoke` 的 input，取不到历史	它要从 `{ configurable: { sessionId } }` 传，属于运行时上下文
多轮里用户说"它支持哪些格式"，召回全是噪声	检索前必须做 Query 改写，用 `createHistoryAwareRetriever` 补全指代
把对话历史也拼进去做向量检索	历史只用于「改写 query」和「注入 Prompt」，不要拿去检索，会严重稀释语义
同一 sessionId 并发两个请求，历史互相覆盖	加分布式锁或乐观版本号

Agent 与生产化

❌ 坑	✅ 解决方案
Agent 偶发无限循环，一次请求烧掉几万 token	`maxIterations` 必须设（5 以内）；工具数量控制在 10 个以内
工具描述写得笼统，模型该调的不调、不该调的乱调	`description` 就是 Agent 的源码，写清"什么时候该调、什么时候不该调"，参数用 `.describe()` 说明
Agent 直接执行了退款操作	工具分级：写操作返回"待确认"结构体，由前端弹框、用户确认后才真正执行
固定流程也上了 Agent，延迟和成本翻几倍且不稳定	流程确定就用 LCEL 写死；需要动态决策但步骤可枚举用 LangGraph
Callback 里直接写数据库，接口 P99 飙升	Callback 默认同步阻塞，只能做内存累加或投队列
Callback 内部抛异常把整条链带崩	handler 里自行 try-catch，监控逻辑不能影响主流程
`MetricsHandler` 的 Map 只 set 不 delete，内存缓慢泄漏	`handleLLMEnd` / `handleLLMError` 里都要 `delete(runId)`
开了 `LANGCHAIN_TRACING_V2` 后被安全团队叫停	LangSmith 是 SaaS 会上传 Prompt 与文档内容；企业内网走自定义 Callback 上报或 OpenTelemetry
SSE 接口在本地正常，上线后前端收不到增量	Nginx 缓冲所致：响应头加 `X-Accel-Buffering: no`，Nginx 侧 `proxy_buffering off`
用户关页面后后端还在生成，白花钱	监听 `req.on("close")`，把 `AbortSignal` 传进 `invoke` / `streamEvents`
流式中途报错，前端一直转圈	流内 try-catch，推一条 `type: error` 再 `res.end()`
文档更新了但答案还是旧的	LLM 缓存 key 没带知识库版本号；灌库完成后要 bump 版本使缓存失效
批量灌库跑到一半抛错，整批中断	`batch(inputs, { maxConcurrency }, { returnExceptions: true })`，单条失败不影响全局
并发打满触发供应商限流	`maxConcurrency` 按 RPM 限额倒推（通常 3~10）；Embedding 用 `embedDocuments` 一次传多条而不是并发多请求

📌 当日复盘任务

框架定位与组件

[ ] 能用一句话说清 LangChain 的定位：LLM 应用的标准接口层 + 编排 DSL，不提供模型也不提供向量库
[ ] 能按数据流向默写八大核心组件及各自的输入输出契约
[ ] 说清 Document 的两个字段，并举出三个必须写进 metadata 的工程化字段（来源、chunkId、tenantId）
[ ] 讲清新版包结构（@langchain/core / provider / community / langchain）以及双实例 Runnable 报错的成因与排查命令
[ ] 说清 LLM 与 ChatModel 的差异，以及四类消息角色的用途和 ToolMessage 的 tool_call_id 约束

LCEL 编排

[ ] 能解释"组合即继承能力"：说出 LCEL 链自动获得的五种能力
[ ] 默写 RunnableSequence / RunnableParallel / RunnablePassthrough.assign / RunnableLambda / RunnableBranch 的用法
[ ] 说清 stream 与 streamEvents 的区别，并解释"中间有非流式节点会堵住流"的原因
[ ] 说清老 Chain 被废弃的四条根本原因，并能把一段 LLMChain 代码改写成 LCEL
[ ] 说清 withRetry 与模型 maxRetries 相乘的坑，以及哪些错误该重试、哪些该降级
[ ] 用 configurableFields 实现一次"同一条链运行时切模型"

检索与切片

[ ] 画出 RecursiveCharacterTextSplitter 的递归降级流程，解释中文为什么会掉到逐字符硬切
[ ] 说清 chunkSize 的单位、chunkOverlap 的经验值，以及如何精确控 token
[ ] 说清 VectorStore 与 Retriever 的分层关系，以及 asRetriever() 的抽象价值
[ ] 手写一个继承 BaseRetriever 的混合检索器，验证 LangSmith 里仍能看到独立 retriever 节点
[ ] 说清五个官方高级 Retriever 的原理与代价，并解释 ParentDocumentRetriever 解决的核心矛盾
[ ] 解释"权限过滤必须下推到检索层"的原因

多轮与记忆

[ ] 说清旧 Memory API 已废弃，并能用 RunnableWithMessageHistory 跑通一次多轮
[ ] 解释 sessionId 为什么从 configurable 传而不是从 input 传
[ ] 对比三种历史管理策略的成本与信息损失，给出生产组合方案
[ ] 实现一次"成对丢弃 + 滚动摘要 + token 上限"的历史压缩，并覆盖单条超长的边界
[ ] 说清 Query 改写（createHistoryAwareRetriever）解决的问题及其副作用

Agent 与生产化

[ ] 画出 Agent 的工具调用循环，说明 tool_calls → ToolMessage 的闭环
[ ] 给出"该用 Agent / 该用 LangGraph / 该用固定 LCEL 链"的三段判断标准
[ ] 写一个 Callback 统计 token、耗时、召回条数，并说明为什么不能在里面写库
[ ] 跑通一个 SSE 流式接口，覆盖 Nginx 缓冲、客户端中断、流内报错、心跳四个坑
[ ] 说出三层成本优化手段，并解释 LLM 缓存 key 为什么必须带知识库版本号
[ ] 能复述"框架薄、自研厚"的选型表：哪些环节用框架、哪些必须自研、理由是什么

面试输出与知识串联

[ ] 把题 2（LCEL vs 老 Chain）、题 4（Retriever 抽象）、题 9（框架封装代价）的 90 秒口述答案各练一遍，确保不看稿
[ ] 把 Day25 手写的 Python RAG 管线用 LCEL 重新编排一遍，逐段对照：哪些代码被框架省掉了、哪些反而变复杂了
[ ] 做一次双端翻译练习：把本日任意一条 TS 链改写成 Python（|、snake_case、ainvoke、astream_events(version="v2")）
[ ] 串联 Day23（Prompt 与流式输出）、Day24（向量库与检索）、Day25（RAG 全链路）：说清本日的 Prompts / VectorStore / Retriever 分别封装了它们的哪一层
[ ] 准备一个属于你自己的框架选型故事：为什么引入（或弃用）LangChain → 遇到的具体问题 → 最终的分层方案 → 收益数字
[ ] 准备一个「你踩过但别人想不到」的坑（推荐：双实例 Runnable 报错、withRetry 与 maxRetries 相乘放大、流中间插了非流式节点导致首字延迟暴涨）
[ ] 预习 Day27：Agent 智能体基础 & Function Calling —— 从「固定链路」走向「模型自己决定下一步」，重点看工具调用协议（tool_calls / tool_call_id 的消息序列）、工具描述与 JSON Schema 怎么写、ReAct 循环的最小实现，以及无限循环与成本怎么加闸

Day26 LangChain 框架开发 ​

📖 基础知识讲解【基础原理 + 高级深挖知识点】 ​

基础原理 ​

1. LangChain 到底解决什么问题 ​

2. 八大核心组件与它们的输入输出契约 ​

3. 新版包结构：为什么 import 路径和教程不一样 ​

4. Models：LLM 与 ChatModel 的差异 ​

5. Prompts：模板、消息模板与 FewShot ​

6. DocumentLoaders：把异构数据源归一成 Document ​

7. TextSplitters：递归切分的分隔符优先级 ​

8. VectorStores 与 Retrievers 的分层关系 ​

9. Chains 的演进：从 LLMChain 到 LCEL ​

10. LCEL：管道符的本质 ​

11. 三个最常用的 Runnable 原语 ​

12. OutputParsers：从字符串到结构化 ​

13. Memory 的现状：旧 API 已废弃 ​

14. 预制 Chain：createRetrievalChain 全家桶 ​

15. Python 版与 JS 版：同一套设计，两套手感 ​

✨ 高级深挖知识点 ​

1. Runnable 接口全貌：一张表吃透 LCEL ​

2. RunnableBranch：按意图路由到不同链路 ​

3. 流式：stream 与 streamEvents 的本质区别 ​

4. 流式 + 引用：先给引用还是先给正文 ​

5. 容错：withFallbacks + withRetry 的正确层级 ​

6. 自定义 Retriever：把自研检索塞进框架 ​

7. 官方高级 Retriever：什么时候真的该用 ​

8. Tools 与 Agent：让模型自己决定调什么 ​

9. Agent 的真实代价与 LangGraph ​

10. Callback：可观测性的统一入口 ​

11. LangSmith：链路追踪落地 ​

12. 成本与缓存 ​

13. 配置动态化：一条链支持多模型 ​

14. 并发与批量：离线评测的提速手段 ​

15. 与 FastAPI / Node 服务对接：SSE 流式接口 ​

16. 选型结论：框架薄、自研厚 ​

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】 ​

题 1：LangChain 是干什么的？核心组件有哪些？ ​

题 2：LCEL 和老的 Chain（LLMChain、ConversationChain）有什么区别？为什么要废弃老写法？ ​

题 3：RecursiveCharacterTextSplitter 的分隔符顺序是怎么回事？中文为什么要特殊处理？ ​

题 4：Retriever 这个抽象的价值在哪？怎么自定义？ ​

题 5：LangChain 的 Memory 怎么实现的？多轮对话 token 爆炸怎么办？ ​

题 6：多轮对话里用户说"它支持哪些格式"，你怎么保证检索准确？ ​

题 7：LangChain 的链路怎么做可观测？出了问题怎么排查？ ​

题 8：什么时候该用 Agent，什么时候不该用？ ​

题 9：LangChain 这层封装有什么代价？你们怎么权衡的？ ​

题 10：怎么给 LangChain 应用做成本控制？ ​

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】 ​

题 1：用 LCEL 写一条最小可用的 RAG 链 ​

题 2：实现一个混合检索的自定义 Retriever ​

题 3：实现带摘要压缩与 token 上限的会话历史 ​

题 4：给链路加上重试、降级、超时、可观测四件套 ​

题 5：用 streamEvents 实现「先出来源、再流正文」的 SSE 接口 ​

⚠️ 当日踩坑实录 ​

📌 当日复盘任务 ​

Day26 LangChain 框架开发

📖 基础知识讲解【基础原理 + 高级深挖知识点】

基础原理

1. LangChain 到底解决什么问题

2. 八大核心组件与它们的输入输出契约

3. 新版包结构：为什么 `import` 路径和教程不一样

4. Models：LLM 与 ChatModel 的差异

5. Prompts：模板、消息模板与 FewShot

6. DocumentLoaders：把异构数据源归一成 Document

7. TextSplitters：递归切分的分隔符优先级

8. VectorStores 与 Retrievers 的分层关系

9. Chains 的演进：从 LLMChain 到 LCEL

10. LCEL：管道符的本质

11. 三个最常用的 Runnable 原语

12. OutputParsers：从字符串到结构化

13. Memory 的现状：旧 API 已废弃

14. 预制 Chain：`createRetrievalChain` 全家桶

15. Python 版与 JS 版：同一套设计，两套手感

✨ 高级深挖知识点

1. Runnable 接口全貌：一张表吃透 LCEL

2. RunnableBranch：按意图路由到不同链路

3. 流式：`stream` 与 `streamEvents` 的本质区别

4. 流式 + 引用：先给引用还是先给正文

5. 容错：`withFallbacks` + `withRetry` 的正确层级

6. 自定义 Retriever：把自研检索塞进框架

7. 官方高级 Retriever：什么时候真的该用

8. Tools 与 Agent：让模型自己决定调什么

9. Agent 的真实代价与 LangGraph

10. Callback：可观测性的统一入口

11. LangSmith：链路追踪落地

12. 成本与缓存

13. 配置动态化：一条链支持多模型

14. 并发与批量：离线评测的提速手段

15. 与 FastAPI / Node 服务对接：SSE 流式接口

16. 选型结论：框架薄、自研厚

❓ 高频面试题【基础简答 + 连环追问链 + 精简口述答案 + 项目关联话术】

题 1：LangChain 是干什么的？核心组件有哪些？

题 2：LCEL 和老的 Chain（LLMChain、ConversationChain）有什么区别？为什么要废弃老写法？

题 3：`RecursiveCharacterTextSplitter` 的分隔符顺序是怎么回事？中文为什么要特殊处理？

题 4：Retriever 这个抽象的价值在哪？怎么自定义？

题 5：LangChain 的 Memory 怎么实现的？多轮对话 token 爆炸怎么办？

题 6：多轮对话里用户说"它支持哪些格式"，你怎么保证检索准确？

题 7：LangChain 的链路怎么做可观测？出了问题怎么排查？

题 8：什么时候该用 Agent，什么时候不该用？

题 9：LangChain 这层封装有什么代价？你们怎么权衡的？

题 10：怎么给 LangChain 应用做成本控制？

✍️ 手撕代码【核心实现 + 边界 Case 自测清单 + 变种拓展题】

题 1：用 LCEL 写一条最小可用的 RAG 链

题 2：实现一个混合检索的自定义 Retriever

题 3：实现带摘要压缩与 token 上限的会话历史

题 4：给链路加上重试、降级、超时、可观测四件套

题 5：用 `streamEvents` 实现「先出来源、再流正文」的 SSE 接口

⚠️ 当日踩坑实录

📌 当日复盘任务