LangChain 与 LangGraph 知识点详解

LangChain 的核心抽象有哪些？解决了什么问题？

核心抽象

LangChain 把「用 LLM 构建应用」拆成一组可组合的抽象单元：

解决了什么问题

1. 统一接口：换模型（OpenAI → Claude → 本地）只需改配置，业务代码不变。
2. 组合编排：把 prompt、模型、解析器串成一条链，减少样板代码。
3. 生态复用：loader/splitter/retriever 等常用组件开箱即用。
4. 可观测：统一 callback/tracing，方便调试与监控。

核心思想

• 「一切皆 Runnable」：每个组件实现统一的 invoke/stream/batch 接口，可用管道符自由组合。

Prompt Template 是什么？有哪些类型？

定义

Prompt Template：把 prompt 中变化的部分用占位变量表示，运行时填充，从而复用 prompt 结构。它是「提示工程」的工程化封装。

from langchain_core.prompts import PromptTemplate
t = PromptTemplate.from_template("把下面内容翻译成{target_lang}：{text}")
t.format(target_lang="英文", text="你好")
# -> "把下面内容翻译成英文：你好"

类型

关键特性

• partial（部分填充）：先填一部分变量，剩余运行时再填。
• 变量校验：缺变量时报错，避免拼出残缺 prompt。
• 生产中用 ChatPromptTemplate.from_messages([("system", ...), ("human", ...)]) 最普遍。

LLM 和 ChatModel 有什么区别？

区别

现状

• 现代模型基本都是 ChatModel（messages 接口）；LLM 是兼容旧「纯补全」模型的抽象。
• 二者都实现了统一的 Runnable 接口（invoke/stream/batch），可互换进 chain。

示例

# ChatModel：输入输出都是消息
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")
resp = model.invoke([("human", "你好")])   # AIMessage

# LLM：纯字符串
prompt = "给这家公司起个名字"
resp = llm.invoke(prompt)   # str

output parser 是什么？如何做结构化输出？

定义

Output Parser：把 LLM 的文本输出解析成结构化对象（dict/Pydantic 模型），是「非结构化 → 结构化」的桥梁。

如何做结构化输出

1. 提示约束：在 prompt 中给出期望格式（如「只输出 JSON」）。
2. 模型生成：LLM 输出一段文本。
3. 解析器解析：parser 把文本转成目标结构，失败则抛错或重试。

from langchain_core.output_parsers import JsonOutputParser
parser = JsonOutputParser()
parser.parse('{"name": "张三", "age": 30}')  # -> {"name": "张三", "age": 30}

常见 parser

进阶：结构化输出（function calling）

• 现代更可靠做法：用工具的 JSON Schema 约束输出（with_structured_output），让模型直接产出合法结构，而不是「生成文本再解析」。

chain 是如何组合的？有哪些组合方式？

组合方式

传统 Chain 与 LCEL

• 旧版用 SequentialChain、LLMChain 等类组合；新版推荐用 LCEL 管道符 | 组合 Runnable，更简洁。

# LCEL 组合 prompt -> model -> parser
chain = prompt | model | StrOutputParser()
result = chain.invoke({"question": "..."})

原理

• chain 本质是一个有向图（DAG）：节点是 Runnable，边是数据流。
• 组合后得到的仍是 Runnable，可继续嵌套、复用。

LCEL（LangChain Expression Language）的引入动机是什么？

动机

早期 LangChain 用 LLMChain、SequentialChain 等类继承体系组合组件，带来问题：

1. 组合僵硬：每个组合都要新建类，嵌套复杂。
2. 接口不统一：不同组件调用方法五花八门（predict、run、generate）。
3. 难以复用：子链难以抽出来独立复用。
4. 难做异步/流式：流式、批量、回退等横切能力难以统一实现。

LCEL 是什么

• 声明式组合语法：用 |（管道符）把 Runnable 串起来，类似 Unix 管道 / RxJS。
• 统一接口：invoke / stream / batch / ainvoke / astream / abatch。
• 自动获得：流式、异步、批处理、重试、回退、并行等能力。

chain = prompt | model | parser     # 声明式，可读性高

结果

• 组合从「继承 + 配置对象」变成「函数式组合」，更简单、更可测试、更易扩展。

管道符 | 和 Runnable 是什么关系？

关系

• | 是 LCEL 的组合操作符，Runnable 是被组合的单元。
• 所有组件（prompt/model/parser/retriever…）都实现 Runnable 协议，因此都能用 | 串起来。

chain = prompt | model | parser
# 等价于把三个 Runnable 依次连接，前一个的输出作为后一个的输入

Runnable 协议

invoke(input) -> output          # 同步执行
ainvoke(input) -> output         # 异步执行
stream(input) -> Iterator        # 流式
batch(inputs) -> list            # 批量

原理

• a | b 内部生成一个 RunnableSequence，执行时把 a 的输出传给 b。
• 每个 Runnable 自带配置传递（config 含 callbacks、metadata、tags），沿链自动传播。

好处

• 任意两个满足协议的组件都能自由组合；组合结果仍是 Runnable，可继续 |。

RunnableParallel、RunnableLambda、RunnablePassthrough 分别是什么？

RunnableParallel

• 并行执行多个 Runnable，输入分发给每个分支，结果合并成 dict。

chain = RunnableParallel(joke=prompt | model, poem=prompt2 | model)
result = chain.invoke({"topic": "猫"})   # {"joke": ..., "poem": ...}

• 用途：一次请求同时产出多路结果，降低多轮串行延迟。

RunnableLambda

• 把任意 Python 函数包装成 Runnable，使其能进 chain。

from langchain_core.runnables import RunnableLambda
upper = RunnableLambda(lambda x: x.upper())
chain = prompt | model | upper

• 用途：在链中插入自定义处理逻辑（清洗、转换、后处理）。

RunnablePassthrough

• 原样透传输入，或从输入中抽取字段，常用于「保留原始输入供后续使用」。

chain = RunnablePassthrough.assign(answer=prompt | model)
# 输入 {"question": ...} -> 输出 {"question": ..., "answer": ...}

• assign：在透传的同时追加新字段。

with_fallbacks 是什么？如何做容错？

定义

with_fallbacks：给一个 Runnable 配置多个备用实现，主实现失败（异常/超时）时自动降级到下一个，是链级容错机制。

robust = primary_chain.with_fallbacks([backup_chain, last_resort_chain])

工作机制

1. 按顺序尝试主链 → 备用链1 → 备用链2。
2. 触发条件：抛出指定的异常类型（如 RateLimitError、TimeoutError）。
3. 某个成功即返回，全部失败才抛错。

from langchain_core.runnables import RunnableLambda
chain = main.with_fallbacks(
    [backup],
    exceptions_to_handle=(TimeoutError,),
)

容错场景

注意

• 备用链要保证输入输出接口一致；fallback 是「整链替换」，不是「单节点重试」（重试用 with_retry）。

LangChain 0.1、0.2、1.0 版本有哪些变化？

版本演进

关键趋势

1. 单体包 → 模块化包：核心抽象在 langchain-core，各供应商集成在 langchain-<vendor>。
2. LCEL / Runnable 成为统一接口：淘汰旧的 LLMChain 类组合。
3. LangGraph 接管 Agent/状态编排：复杂流程、Agent 用 LangGraph，LangChain 提供组件与轻量链。
4. 1.0 语义：明确「不追求功能堆叠，追求 API 稳定、可长期依赖」。

意义

• 理解版本演进能体现「对生态架构有整体把握」，面试常考「LangChain 与 LangGraph 的关系」。

memory（BufferMemory）是如何工作的？有哪些类型？

工作原理

• Memory 负责把历史对话状态存入/取出，让链在多次调用间「记住」上下文。
• 调用前：把历史消息注入 prompt；调用后：把本轮交互写回 memory。

用户: 我叫小明
模型: 你好小明
用户: 我叫什么？   <- memory 里有「我叫小明」，能答出

类型

注意

• 传统 ConversationBufferMemory 已弃用，新版推荐用 RunnableWithMessageHistory + BaseChatMessageHistory 管理对话历史。

bind_tools 是什么？如何绑定工具？

定义

bind_tools：把工具 schema（名称、描述、参数）绑定到 ChatModel 上，使模型在下一次调用时能返回 tool call（结构化调用指令）。

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    return "北京今天 18°C"

model_with_tools = model.bind_tools([get_weather])
ai_msg = model_with_tools.invoke("北京天气怎么样")
# ai_msg.tool_calls -> [{"name": "get_weather", "args": {"city": "北京"}}]

原理

1. 工具被序列化成 JSON Schema 传给模型。
2. 模型判断需要工具时，返回 tool_calls（含函数名 + 参数）。
3. 宿主执行工具，结果以 tool role 回传，模型继续生成。

关键点

• 绑定后模型并不会真正执行工具，只是输出调用指令。
• 工具的 description（含 docstring）质量决定模型选工具的准确率。
• 多个工具一次绑定，模型可做「选择 + 参数抽取」，支持并行调用。

retriever 抽象是什么？如何自定义？

定义

Retriever：统一的「输入 query，输出相关文档列表」的接口，是 RAG 中「检索」环节的抽象。

Retriever.get_relevant_documents(query) -> List[Document]

• 屏蔽底层实现（向量检索 / BM25 / 图检索），让上层 chain 无感切换。

如何自定义

实现 BaseRetriever，重写 _get_relevant_documents：

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document

class MyRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str, **kwargs):
        # 自定义检索逻辑：向量库 / 数据库 / 远程 API
        return [Document(page_content="命中内容", metadata={"source": "..."})]

常见 retriever

vector store 抽象是什么？如何接入？

定义

Vector Store：负责向量存储与相似度检索的抽象，RAG 的「知识库底座」。

add_documents(docs) / add_texts(texts)     # 写入（先 embedding 再入库）
similarity_search(query, k)                # 按相似度检索 top-k
similarity_search_with_score(query, k)     # 带分数检索
as_retriever()                             # 转成 Retriever 接进 chain

如何接入

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
chain = retriever | prompt | model | parser

常见实现

关键点

• 写入时自动调 embedding 向量化；检索时对 query 向量化后做 ANN 检索。
• 切换 vector store 只需换实现类，as_retriever() 后上层代码不变。

text splitter 有哪些类型？如何选择？

类型

如何选择

• 通用文本 → RecursiveCharacterTextSplitter。
• Markdown/文档 → MarkdownHeaderTextSplitter（按标题）。
• 代码 → 对应语言的 CodeSplitter。
• 要精确 token 预算 → TokenTextSplitter。
• 追求语义完整 → SemanticChunker（慢）。

关键参数

• chunk_size：块大小（常见 256~1024 token）。
• chunk_overlap：块间重叠（10%~20%），避免语义被切断。
• 原则：块过大检索噪声多、块过小语义破碎。

document loader 有哪些类型？

定义

Document Loader：把各类数据源加载为 Document 对象（page_content + metadata），是 RAG 的数据入口。

类型

通用加载方式

from langchain_community.document_loaders import DirectoryLoader, TextLoader

loader = DirectoryLoader("./docs", glob="**/*.md", loader_cls=TextLoader)
docs = loader.load()   # -> List[Document]

关键点

• 核心产物是 Document 列表（正文 + 元数据），后续接 splitter → embedding → vector store。
• 复杂格式（PDF 表格、扫描件）常用 Unstructured 生态（OCR/版面解析）。

embedding 抽象是什么？如何接入不同的 embedding？

定义

Embedding 抽象：统一的「文本 → 向量」接口，屏蔽不同 embedding 模型/服务的差异。

embed_documents(texts) -> List[vector]   # 文档侧（离线建库）
embed_query(text)      -> vector         # 查询侧（在线检索）

如何接入不同 embedding

from langchain_openai import OpenAIEmbeddings   # OpenAI
# from langchain_huggingface import HuggingFaceEmbeddings  # 本地 BGE 等
# from langchain_google_genai import GoogleGenerativeAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vecs = embeddings.embed_documents(["你好", "世界"])

• 换服务只改实现类，vectorstore 与上层代码不用改。

关键点

• 文档与查询分两接口：不对称检索（query/doc 用不同 prompt 或模型）得以支持。
• 选择看：语言能力（中文选 BGE）、维度、检索专用性、成本/延迟（API vs 本地）。
• embedding 维度必须与向量库建索引时一致。

callbacks 和 tracing 如何工作？

callbacks

• 回调钩子：在链/模型执行的生命周期事件上挂自定义函数，实现日志、监控、计费等横切逻辑。

on_llm_start / on_llm_end        # 模型调用前后
on_chain_start / on_chain_end    # 链执行前后
on_tool_start / on_tool_end      # 工具调用前后
on_retriever_start / on_retriever_end

from langchain_core.callbacks import BaseCallbackHandler

class MyHandler(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        print("token 用量:", response.llm_output)

tracing

• tracing 是 callbacks 的高级应用：把整条执行轨迹（trace）与子步骤（span）记录下来，形成调用树。
• 通过 config={"callbacks": [...]} 或环境变量（LANGCHAIN_TRACING_V2=true）把轨迹上报到 LangSmith / Langfuse。

用途

• 调试（看每步输入输出）、评估（轨迹打分）、成本分析、回归测试。

streaming（astream）如何实现？

概念

• streaming：不等待完整结果，而是逐 token/逐块返回，让前端实时显示，降低感知延迟。
• astream 是异步流式版本，astream_events 提供更细粒度的事件流。

实现

chain = prompt | model | StrOutputParser()

# 同步流式
for chunk in chain.stream({"topic": "AI"}):
    print(chunk, end="")     # 逐 token 打印

# 异步流式
async for chunk in chain.astream({"topic": "AI"}):
    print(chunk, end="")

关键点

• LCEL 组合的链自动支持流式：上游 Runnable 的流式会「穿透」整条链（只要下游也支持）。
• token 流（模型）与工具调用流（增量解析 JSON）是两层。
• 复杂场景用 astream_events 获取 on_llm_start/on_token/on_tool_start/... 事件序列。

注意

• StrOutputParser 会把 token 逐块透出；某些 parser（如 JSON）需等完整输出才能解析，会破坏纯 token 流。

batch 和 abatch 有什么区别？

区别

用法

inputs = [{"question": "1+1"}, {"question": "2+2"}]

results = chain.batch(inputs)           # 同步，返回 [r1, r2]
results = await chain.abatch(inputs)    # 异步并发

关键点

• batch 会把多个输入并发执行（受 max_concurrency 限制），比循环 invoke 更快。
• abatch 是 batch 的异步版本，适合在 asyncio 服务里高并发调用。
• 三者都是 Runnable 统一接口的一部分：invoke（单条）、batch/abatch（多条）、stream/astream（流式）。

RunnableWithMessageHistory 是什么？

定义

• 在链上自动管理对话历史：调用前从存储加载历史消息，调用后把本轮交互写回，实现多轮会话。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

chain = prompt | model
with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: ChatMessageHistory(),
    input_messages_key="input",
    history_messages_key="history",
)
resp = with_history.invoke(
    {"input": "我叫小明"},
    config={"configurable": {"session_id": "user-1"}},
)

工作原理

1. 用 session_id 定位某个会话的历史存储（get_session_history 返回）。
2. 取出历史，作为 history 变量注入 prompt（配合 MessagesPlaceholder）。
3. 执行链，然后把本轮输入 + 输出写回历史。

关键点

• session_id 是会话隔离的键，多用户靠它区分。
• 底层 BaseChatMessageHistory 可接内存 / Redis / 数据库。
• 这是新版推荐的多轮记忆方案（替代旧 ConversationBufferMemory）。

LangChain 的 caching 如何实现？

为什么缓存

• 相同请求避免重复调 LLM，降成本、降延迟。

实现方式

1. 内存缓存（开发用）：

from langchain_core.globals import set_llm_cache
from langchain_core.caches import InMemoryCache
set_llm_cache(InMemoryCache())

2. 持久化缓存（生产用）：

from langchain_community.cache import SQLiteCache, RedisCache
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

3. 语义缓存（GPTCache / SemanticCache）：按语义相似度命中，而非逐字相等——问法不同但意思相同也能命中。

缓存键

• 默认以「模型 + prompt」为键；相同 prompt 命中缓存直接返回，不调模型。

注意

• 缓存适合确定性、重复性高的查询（如 FAQ）；对需要实时/随机性的场景要谨慎。
• 语义缓存命中率更高，但可能带来语义偏差，需阈值控制。

LangChain 有哪些优缺点？

优点

1. 抽象统一：模型/prompt/retriever 等统一接口，快速搭建原型。
2. 生态丰富：loader/splitter/tool/integration 众多，开箱即用。
3. 可组合（LCEL）：管道符声明式组合，自动获得流式/异步/批处理。
4. 可观测：内置 callbacks，对接 LangSmith 等 tracing。
5. 社区大：资料、集成、教程多，招聘市场认可度高。

缺点

1. 抽象过度：层层封装，隐藏细节，排查问题难、学习曲线陡。
2. API 变动大：0.1/0.2/1.0 频繁破坏性变更，升级成本高。
3. 复杂流程力不从心：链式 DAG 难以表达循环、条件分支、动态路由。
4. 性能与体积：抽象层带来一定开销，依赖重。
5. 调试难：报错栈深、黑盒，难以定位问题根源。

一句话总结

• 「原型快、生产要谨慎」：快速验证很强，深度定制/复杂 Agent 时反而要绕开抽象。

为什么 LangChain 被批评「抽象过度」？

原因

1. 封装层次多：一个「调模型」可能穿过 prompt→model→parser→callback 多层，报错栈极深，难以定位。
2. 隐藏关键细节：开发者看不到底层 prompt 拼装、消息构造、重试逻辑，出了问题「黑盒」。
3. 魔法配置：大量隐式默认（自动工具选择、自动解析），行为不可预期，难以精确控制。
4. API 频繁变动：为了兼容旧抽象不断叠新抽象，概念膨胀（Chain/Agent/AgentExecutor…）。
5. 简单任务被复杂化：只是「一个 prompt + 一个模型」也要引入一堆抽象，收益低、心智负担高。

反方观点（LangChain 的回应）

• 抽象带来可移植性（换模型不改代码）与可组合性（LCEL 复用）。
• 1.0 后收敛 API、LangGraph 专注编排，试图「该简单的地方简单、该复杂的地方可控」。

结论

• 批评的实质是「抽象收益与成本不匹配」：对简单场景过度设计，对复杂场景又不够灵活（才催生 LangGraph）。

LangChain 适用和不适用于哪些场景？

适用场景

1. 快速原型 / PoC：想几天内跑通 RAG、聊天机器人。
2. 标准 RAG 流水线：loader → splitter → embedding → vector store → retriever。
3. 多模型切换需求：需要屏蔽 OpenAI/Claude/本地模型差异。
4. 简单链式应用：prompt → model → parser 的直线流程。
5. 需要现成组件：大量 loader/tool/integration 直接用。

不适用场景

1. 复杂状态机 / 多步 Agent：循环、分支、回溯、人工介入 → 用 LangGraph。
2. 极致性能/延迟敏感：抽象开销、依赖体积是负担。
3. 需要完全可控的底层细节：不想被黑盒封装挡住。
4. 高度定制的工作流：自己写编排更直接。
5. 长期稳定 API 要求：版本变动大（1.0 后改善）。

结论

• 经验法则：「原型/标准 RAG 用 LangChain，复杂 Agent/状态编排用 LangGraph」。

LangGraph 解决什么问题？和 LangChain 的 chain 有什么区别？

解决什么问题

• LangChain 的 chain 是静态 DAG：数据单向流动，难以表达循环、条件分支、动态路由、人工介入、持久化状态。
• LangGraph 用「图 + 状态」建模，天然支持循环、分支、并行、回溯、checkpoint，是 Agent 与复杂工作流的编排层。

区别

关系

• 互补而非替代：LangGraph 内部节点常用 LangChain 组件（模型、工具、retriever）；LangChain 提供积木，LangGraph 负责搭复杂结构。

StateGraph 和 State 是什么？

State

• State：图执行过程中共享的状态对象，通常是 TypedDict / Pydantic，节点通过它读写数据。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 用 reducer 合并
    count: int

StateGraph

• StateGraph：以 State 为中心的图构建器，定义「节点 + 边 + 状态如何更新」。

from langgraph.graph import StateGraph

builder = StateGraph(State)
builder.add_node("node_a", fn_a)
builder.add_node("node_b", fn_b)
builder.add_edge("node_a", "node_b")
builder.set_entry_point("node_a")
graph = builder.compile()

状态更新（reducer）

• 每个节点返回一个状态片段，按 channel 的 reducer 合并进整体 State。
• 默认「后写覆盖前写」；add_messages 这类 reducer 则追加（消息累加）。

关键点

• State 是 LangGraph 的「显式共享内存」，比 chain 的「隐式数据流」更可控、可持久化、可检查。

node 和 edge 是什么？如何定义？

node（节点）

• 图中的一个执行单元，本质是函数：接收当前 State，返回状态更新片段。

def node_a(state: State):
    # 处理逻辑
    return {"count": state["count"] + 1}   # 返回要更新的字段

• 节点可以是普通函数，也可以是 LangChain Runnable / 模型调用 / 工具执行。

edge（边）

• 定义节点间的执行顺序，分两类：

builder.add_edge("node_a", "node_b")      # 普通边
builder.add_conditional_edges("node_a", route_fn, {"x": "node_b", "y": "node_c"})
builder.set_entry_point("node_a")          # 入口
builder.set_finish_point("node_b")         # 出口（可多个）

关键点

• compile() 后得到可 invoke/stream 的图；图 = 节点（做什么）+ 边（何时做/按什么顺序）。
• 条件边用 route_fn(state) -> "下一节点名" 决定路由。

条件边（conditional edge）是什么？

定义

• 条件边：从一个节点出发，根据当前状态/输出动态决定下一个节点，是图「智能路由」的核心。

实现

def route(state) -> str:
    if "需要工具" in state["messages"][-1].content:
        return "tools"
    return "end"

builder.add_conditional_edges(
    "agent",
    route,                                # 路由函数
    {"tools": "tools", "end": END},       # 返回值 -> 目标节点映射
)

工作流程

1. 源节点执行完，调 route(state) 返回一个字符串。
2. 用映射表把该字符串解析成目标节点名。
3. 跳转到对应节点继续执行。

用途

• ReAct Agent 的「继续调工具 vs 结束」判断。
• Supervisor 多 Agent 的「该派给哪个 worker」路由。
• 异常/兜底分支：依状态走正常流程或降级流程。

LangGraph 如何支持循环？

核心机制

• LangGraph 的图允许有环：节点 A → B → A 形成循环，靠条件边 + 终止条件控制退出。

agent --条件边--> tools --边--> agent   # 循环
agent --条件边--> END                    # 终止

ReAct 循环示例

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")       # 工具执行完回到 agent，形成循环

def should_continue(state):
    last = state["messages"][-1]
    return "tools" if last.tool_calls else "end"

终止保障

• 循环必须有退出条件，否则死循环：
  1. 条件边返回 END。
  2. 设置 recursion_limit（最大步数）强制终止。
  3. 节点内检测重复/超限主动结束。

graph = builder.compile(checkpointer=..., interrupt_before=[...])
result = graph.invoke(input, config={"recursion_limit": 25})

checkpoint 持久化是什么？如何实现？

定义

• checkpoint：图执行到每个节点后保存的 State 快照，用于恢复、续跑、人工介入、时间旅行。

如何实现

1. 编译时挂 checkpointer：

from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

2. 调用时给 thread_id 标识会话：

config = {"configurable": {"thread_id": "user-123"}}
graph.invoke(input, config)   # 每次调用自动存 checkpoint

作用

生产 checkpointer

• 内存 MemorySaver（开发）、SqliteSaver、PostgresSaver（生产持久化）。

human-in-the-loop（interrupt）如何实现？

定义

• human-in-the-loop（HITL）：在关键节点暂停执行，等待人工批准/补充输入后，再 resume 继续。

实现

graph = builder.compile(checkpointer=MemorySaver())

# 在 "tools" 节点执行前暂停
graph = builder.compile(checkpointer=MemorySaver(), interrupt_before=["tools"])

result = graph.invoke(input, config)   # 执行到 tools 前暂停
# ... 人工审查 ...
graph.invoke(None, config)             # 传 None 表示 resume，继续执行

交互方式

1. interrupt_before / interrupt_after：在指定节点前后暂停。
2. 动态 interrupt：节点内用 interrupt() 函数，把值抛给外部并等待。
3. 人工可修改 State（如改工具参数）后再 resume，也可拒绝操作。

场景

• 高危操作（支付/删除/发消息）前审批；需要用户补充信息；人工接管关键决策。

LangGraph 有哪些 streaming modes？

模式

for chunk in graph.stream(input, stream_mode="updates"):
    print(chunk)   # 每个节点更新

for msg, metadata in graph.stream(input, stream_mode="messages"):
    print(msg.content, end="")   # token 级

选择

• 只想显示最终回答 token → messages。
• 想看每个节点产出/进度 → updates。
• 需要精确到节点内部自定义事件 → custom。

关键点

• 一个 stream 可同时传多个模式 stream_mode=["updates", "messages"]。
• 流式 + checkpoint 结合，可实现「边执行边持久化」。

subgraph 如何复用？

定义

• subgraph：把一张图封装成节点嵌入另一张图，实现图级别的模块化与复用。

subgraph_builder = StateGraph(SubState)
# ... 定义子图
subgraph = subgraph_builder.compile()

main_builder = StateGraph(MainState)
main_builder.add_node("sub", subgraph)   # 子图作为节点
main_builder.add_edge(START, "sub")

复用方式

1. 直接作为节点：把编译好的图传给 add_node，主图节点执行时跑完整子图。
2. 状态映射：子图 State 与主图 State 字段不同时，用 add_node(..., input=..., output=...) 做字段映射/转换。
3. 父图访问子图状态：通过 get_state 等 API 查看子图内部状态。

场景

• 多 Agent 中把每个 agent 逻辑封装成子图；复杂流程按模块拆分、独立测试、多处复用。

注意

• 子图需要自己的 State 定义；checkpoint 可下钻到子图（namespace 嵌套）。

LangGraph 如何实现多 Agent（supervisor/swarm）？

supervisor 模式

• 中央 supervisor agent 用条件边把任务路由给多个 worker 节点，worker 完成后回到 supervisor 汇总。

builder.add_node("supervisor", supervisor_fn)
builder.add_node("researcher", researcher_fn)
builder.add_node("coder", coder_fn)
builder.add_conditional_edges("supervisor", route, {
    "researcher": "researcher", "coder": "coder", "FINISH": END})
builder.add_edge("researcher", "supervisor")   # 完成后回到 supervisor
builder.add_edge("coder", "supervisor")

• supervisor 输出「next 节点」的指令（通常用 tool call 或结构化输出解析）。

swarm 模式

• 去中心化：多个 agent 直接 handoff（交接），无固定 supervisor；谁适合谁接手。
• 实现：每个 agent 可输出「转交给谁」的指令，条件边据此跳转。

对比

LangGraph 的 memory（checkpointer + Store）如何设计？

两类记忆

checkpointer（短期）

• 每个 thread_id 保存一份 State 快照，多轮调用复用，实现会话记忆。

graph.compile(checkpointer=PostgresSaver(...))
config = {"configurable": {"thread_id": "user-1"}}

Store（长期）

• BaseStore：跨会话的键值存储，支持命名空间与语义检索，用于长期记忆/跨用户知识。

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
store.put(("memories",), "u1", {"data": "用户偏好 JSON"})

• 节点内通过 store 读写；生产可接 Postgres、Redis 等。

分工

• 短期：同一 thread_id 的上下文连续性 → checkpointer。
• 长期：跨会话、可被检索复用的知识 → Store（可做语义搜索）。

LangGraph 的 pregel 执行模型是什么？

定义

• Pregel：源自 Google 的图计算模型（BSP 批量同步并行），LangGraph 借鉴它做状态图的步进执行。

核心思想

每轮 superstep：
  1. 找到所有「待执行」的节点（有输入就绪）
  2. 并行执行这些节点
  3. 收集它们的 State 更新，合并回整体 State
  4. 进入下一轮，直到无节点可执行

特点

意义

• 让 LangGraph 能表达循环、并行、动态路由，同时保持「每步状态可检查、可持久化」。

time-travel 是什么？如何实现？

定义

• time-travel：回到某个历史 checkpoint，查看或从那里重新执行/分支，用于调试与「如果当初…会怎样」。

实现

1. 拿到历史状态快照列表：

states = graph.get_state_history(config)   # 按时间倒序

2. 查看某个历史状态：

snapshot = states[3]
print(snapshot.values)   # 那个时刻的 State

3. 从历史状态恢复重跑：

graph.invoke(None, {"configurable": {"thread_id": "t1", "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"]}})

4. 或 fork 新分支：换一个 checkpoint_ns / thread_id 从该点分叉，保留原轨迹。

用途

• 调试「哪一步开始跑偏」；回放复现问题；A/B 不同后续策略。

LangGraph 和旧的 AgentExecutor 有什么区别？

AgentExecutor（旧 LangChain）

• 封装好的一体化 Agent 运行时：内部硬编码了「模型 → 工具 → 循环」的逻辑。
• 问题：不透明、难定制——想改循环逻辑、加人工介入、做持久化都要 hack。

LangGraph（新）

• 把 Agent 拆成显式的节点 + 边，循环、路由、工具执行都可见可控。

演进

• create_react_agent（LangGraph 版）取代了旧 create_react_agent/AgentExecutor，底层是显式 ReAct 图，可自由改造。

send 和 MapReduce 是什么？

send

• 动态并行分叉：在条件边/节点里用 Send，把一个节点的多个输出动态派发给多个目标节点，实现类似「扇出」。

from langgraph.types import Send

def continue_to_jokes(state):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

• 用途：一个节点产出 N 个任务，分别并行执行 N 个下游节点。

MapReduce

• Map（映射）：把输入拆成多份，并行处理（每个子任务跑一个子图/节点）。
• Reduce（归约）：把所有子结果合并成最终结果。

问题 -> 拆成子问题 -> [并行处理子问题] -> 汇总 -> 最终答案

关系

• send 是实现 MapReduce「并行 Map + 动态扇出」的机制；MapReduce 是更高层的模式。

@tool 装饰器是什么？如何自定义工具？

定义

• @tool 装饰器把普通 Python 函数转成 LangChain 工具，自动生成 JSON Schema（函数名 + docstring 作为描述 + 参数类型）。

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气，返回温度与天气状况。"""
    # 真实实现：调 API
    return f"{city} 今天 18°C"

工作原理

1. 装饰器读取函数签名（参数名/类型/默认值）与 docstring。
2. 生成工具 schema：name、description、parameters。
3. 工具可 bind_tools 绑定到模型，或直接 .invoke(args) 调用。

自定义要点

• docstring 是给模型的「说明」：写清何时用、返回什么，直接影响调用准确率。
• 参数用类型注解（可加 Annotated 描述、enum 限制取值）。
• 工具内部可调外部 API/数据库；需处理异常并返回结构化结果。

@tool
def delete_file(path: str) -> str:
    """删除指定文件。高危操作，需谨慎调用。"""
    ...

structured tool（Pydantic）是什么？如何定义？

定义

• structured tool：用 Pydantic 模型精确定义工具参数 schema，实现强类型、带校验的工具，模型返回的 args 会被解析并校验成 Pydantic 对象。

from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool

class WeatherInput(BaseModel):
    city: str = Field(description="城市名，如 '北京'")
    date: str = Field(description="日期，YYYY-MM-DD")

def get_weather(city: str, date: str) -> str:
    return f"{city} {date} 天气晴"

tool = StructuredTool.from_function(
    func=get_weather,
    name="get_weather",
    description="查询指定城市指定日期的天气",
    args_schema=WeatherInput,
)

优势

1. 参数校验：类型错误/缺字段会被 Pydantic 校验拦截。
2. schema 精确：字段级 description、enum、约束都进 JSON Schema，模型调用更准。
3. 结构化回传：解析后的 args 是类型化对象，宿主执行更安全。

与 @tool 关系

• @tool 用类型注解自动生成 schema（简单快捷）；structured tool 用显式 Pydantic（更精细，可加复杂校验）。

BaseChatModel 抽象是什么？

定义

• BaseChatModel：LangChain 中「对话模型」的统一抽象基类，屏蔽各家 API 差异，统一暴露 invoke/stream/batch 接口。

核心接口

invoke(messages) -> AIMessage
stream(messages)  -> Iterator[AIMessageChunk]
batch(messages_list) -> list[AIMessage]
bind_tools(tools) -> 绑定工具的模型
with_structured_output(schema) -> 结构化输出模型

输入输出消息类型

意义

• 实现 BaseChatModel 即可接入一个新模型；上层 chain/agent 无需感知具体厂商。
• 所有 ChatModel 都是 Runnable，可直接 | 组合。

自定义

• 继承 BaseChatModel，实现 _generate / _stream（或 _agenerate / _astream），即可包装任意本地/私有模型。

bind 和 config 是什么？如何传递配置？

bind

• bind：在 Runnable 上绑定固定参数，生成一个新的 Runnable；常用于给模型预设 temperature、stop、tools 等。

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)   # 构造时传参
strict_model = model.bind(temperature=0, response_format="json")  # 运行时再绑定

config

• config：调用时传入的运行上下文（RunnableConfig），沿整条链自动传播，包含：

result = chain.invoke(input, config={
    "configurable": {"session_id": "s1"},
    "callbacks": [handler],
    "tags": ["prod"],
})

传递机制

• 每个 Runnable 执行时把 config 逐层透传给下游；子组件可通过 RunnableConfig 读取 configurable 里的自定义值。

LangSmith 是什么？有什么用？

定义

• LangSmith：LangChain 官方推出的 LLM 应用可观测性与评估平台，负责 trace、调试、评估、监控。

用途

接入

• 设环境变量即自动上报：

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxx

• 也可用 SDK 手动记录任意 LLM 调用（不限于 LangChain）。

同类

• 开源替代：Langfuse、Phoenix（Arize）、OpenLLMetry（OpenTelemetry）。

LangServe 是什么？如何部署？

定义

• LangServe：把 LangChain/LangGraph 的 chain/agent 一键部署为 REST API（基于 FastAPI），自带 playground、流式、追踪。

如何部署

1. 定义 chain，用 add_routes 暴露：

from langserve import add_routes
from fastapi import FastAPI

app = FastAPI()
add_routes(app, chain, path="/my_chain")

2. 启动：

langchain app new my-app     # 或手动
uvicorn app:app --host 0.0.0.0 --port 8000

提供的端点

• /my_chain/invoke、/my_chain/stream、/my_chain/batch（对应 Runnable 三接口）。
• 自动生成 OpenAPI 文档与 playground（网页调试 UI）。
• 内置流式（SSE）、token 计数、LangSmith 追踪、configurable 透传。

生产注意

• 用 RunnableConfigurableFields 暴露可配置参数；前面接网关/鉴权/限流；容器化 + 水平扩容。

LangChain 和 LangGraph 如何选型？

选型原则

组合使用

• 多数项目是两者并用：LangChain 提供组件（模型/工具/retriever/prompt），LangGraph 提供编排（状态图、循环、持久化）。

判断依据

• 看是否需要循环/状态/人工介入：需要 → LangGraph；只是直线数据流 → LangChain 足够。

如何用 LangGraph 实现一个 ReAct agent？

结构

START -> agent -> (条件边) -> tools -> agent -> ... -> END

实现

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

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

def agent(state):
    return {"messages": [model.bind_tools(tools).invoke(state["messages"])]}

def tool_node(state):
    last = state["messages"][-1]
    results = [tool_map[tc["name"]].invoke(tc) for tc in last.tool_calls]
    return {"messages": results}

def should_continue(state):
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile()

关键点

• agent 节点：bind_tools 的模型产出含 tool_calls 的 AIMessage。
• tools 节点：执行 tool_calls，把 ToolMessage 结果写回。
• 条件边：判断「继续调工具 or 结束」。
• LangGraph 提供封装 create_react_agent(model, tools)，一行得到标准 ReAct 图，可再自定义。

如何用 LangChain 实现一个 RAG 应用？

完整链路

加载文档 -> 分块 -> embedding -> 存向量库 -> 检索 -> 拼 prompt -> 生成

实现

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. 索引（离线）
docs = TextLoader("knowledge.txt").load()
chunks = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50).split_documents(docs)
vectorstore = Chroma.from_documents(chunks, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 2. 检索 + 生成（在线）
prompt = ChatPromptTemplate.from_messages([
    ("system", "根据以下上下文回答问题：\n{context}"),
    ("human", "{question}"),
])
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt | ChatOpenAI() | StrOutputParser()
)
answer = chain.invoke("什么是 RAG？")

关键点

• 索引离线、检索生成在线两段分离。
• {"context": retriever, "question": RunnablePassthrough()} 用 RunnableParallel 同时产出上下文与问题。
• 检索质量决定 RAG 上限；可加 rerank、query 改写优化。

LangChain/LangGraph 生产部署有哪些注意事项？

注意事项

关键原则

• 状态外置：图/链本身无状态，状态进 checkpointer/数据库，才可水平扩展。
• 可观测先行：没有 tracing，生产出问题无法定位。
• 降级与兜底：模型/工具失败要有降级路径，不让整体挂掉。