LangChain 入门速查表

LangChain 入门速查表

LangChain 入门速查表

Python / Runnable / LCEL / Agent / RAG 一条主线 · 一页纸带走的 LangChain v1 骨架

LangChain v1.3+
LCEL · | 协议
create_agent
RAG 最小闭环
Python ≥ 3.10

🎯 框架定位与版本

LangChain v1 是什么？

LangChain 团队的 LLM 应用编排"主薄"——把模型、Prompt、工具、检索、流式、可观测拼成一站式流水线

包职责初学者是否装

langchain-core
Runnable 协议、Prompt、Messages、BaseTool、输出解析、向量存储抽象
✅ 主包自带

langchain
v1 起精简为 agent 工厂 + 高级 helper(init_chat_model / init_embeddings / create_agent)
✅ 必装

langchain-openai / langchain-anthropic ...
各家模型集成包,按 provider 选用
✅ 至少一个

langchain-classic
已 deprecated 的老 chain / retriever / memory / hub 等旧 API,维护不增功能
⚠️ 老项目才装

langchain-community
第三方 loader/vectorstore/embedding;已 SUNSET (2026-05-22),新项目应绕开
❌ 新项目禁用

langgraph (Checkpointer 来自这里)
会话持久化 InMemorySaver / SqliteSaver / PostgresSaver、thread_id、store
✅ 跑 Agent 必装

一句话：LangChain v1 = Runnable + Agent 工厂;老 chain 迁入 langchain-classic;langchain-community 退场。Python ≥ 3.10。

当前推荐版本(2026-07-21 抓取)

PyPI 元数据,与 docs.langchain.com 交叉印证

langchain 1.3.14(2026-07-16)

langchain-core 1.5.0(2026-07-21)

langchain-openai 1.3.5(2026-07-10)

langchain-anthropic 1.4.8(2026-06-27)

langchain-classic 1.0.8(maintenance mode)

langchain-mcp-adapters 0.3.0(独立包,装 MCP 时再加)

# 推荐: uv(官方 Quickstart 当前写法)
uv add langchain langchain-openai

# pip 也行
pip install -U langchain langchain-openai

API Key 与环境变量

官方 Quickstart 现在用 shell export,不再演示 getpass

Shell 直接 export:export OPENAI_API_KEY=sk-...

init_chat_model 自动读 OPENAI_API_KEY / ANTHROPIC_API_KEY / GOOGLE_API_KEY / AZURE_OPENAI_*

本地协作可用 .env + python-dotenv(官方 Quickstart 不强制)

多 Key 别硬编,在 config["configurable"] 里 with_config(...) 注入

import os
from langchain.chat_models import init_chat_model

model = init_chat_model(
"openai:gpt-4o-mini", # "provider:model" 前缀
temperature=0,
max_retries=6,
)

🧱 核心抽象:Runnable / 模型 / Prompt

Runnable 协议 与 | 管道

LangChain 唯一抽象——所有组件都能 invoke / stream / batch / astream

同步:invoke / batch / stream

异步:ainvoke / abatch / astream / astream_events

| 等价于 RunnableSequence(first, last)

整条链自动继承 sync/async/批/流,不手写胶水

每个 Runnable 支持 RunnableConfig(tags / metadata / callbacks)

from langchain_core.runnables import RunnableLambda
def add1(x): return x + 1
def mul2(x): return x * 2
chain = RunnableLambda(add1) | RunnableLambda(mul2)
chain.invoke(3) # → 8
chain.batch([1, 2, 3]) # → [4,6,8] 并行

核心心智:LangChain 的"组件"≈ Runnable;"组合"≈ 串管道。其它一切围绕这一层。

init_chat_model — 统一模型入口

一句话切 OpenAI / Anthropic / Gemini / Azure / Bedrock

路径:langchain.chat_models.init_chat_model

模型 ID 支持 "provider:model" 前缀推断,或显式 model_provider=

统一 kwargs:temperature / max_tokens / timeout / max_retries / rate_limiter / base_url

configurable_fields="any" + config_prefix → 运行时切模型

# 写法 A: provider:model 前缀
m1 = init_chat_model("openai:gpt-4o-mini")
m2 = init_chat_model("claude-sonnet-4-5-20250929")

# 写法 B: 显式 provider(推荐生产稳定)
m3 = init_chat_model("gpt-4o", model_provider="openai",
temperature=0, timeout=30)

# 写法 C: Azure
m4 = init_chat_model(
"azure_openai:gpt-4o",
azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"])

坑:用 *-latest 别名有漂移风险——生产钉带日期版本(如 claude-haiku-4-5-20251001)。

ChatPromptTemplate

结构化消息模板,变量用 {var} f-string 插值

from_messages([...]):("system", "...") 元组或 SystemMessage(...) 类

MessagesPlaceholder("history", optional=True):动态消息注入(对话历史、检索 Documents)

format(name="Bob", q="...") 或 invoke({...}) 都行

tuple 简写 ("placeholder", "{var}") = MessagesPlaceholder(...)

from langchain_core.prompts import (
ChatPromptTemplate, MessagesPlaceholder)

prompt = ChatPromptTemplate.from_messages([
("system", "你是 {name},用中文简洁回答。"),
MessagesPlaceholder("history", optional=True),
("human", "{question}"),
])
prompt.invoke({"name": "AI 助手",
"history": [("human", "hi")],
"question": "你好"})

输出解析 与 结构化输出

str / pydantic / JSON schema / tool_calls

链式收尾:prompt | llm | StrOutputParser() → 直接拿 str

首选结构化:model.with_structured_output(Schema)(自动选 JSON schema / function calling)

Tool 调用:model.bind_tools([t1, t2]),从 AIMessage.tool_calls 取

旧 PydanticOutputParser / JsonOutputParser 仍可用,但 v1 推荐让模型自己交 schema

from langchain_core.output_parsers import StrOutputParser
from pydantic import BaseModel

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

chain = prompt | model | StrOutputParser() # → str
typed = model.with_structured_output(Movie) # → Movie 实例
print(typed.invoke("介绍一下《盗梦空间》"))

路径注意:v1+ 主推 langchain_core.output_parsers;langchain.output_parsers 已删除。

🤖 Tool 与 Agent

@tool 装饰器

把任意函数变成可被 LLM 调用的工具

docstring = LLM 看到的说明,决定"何时调"

类型注解 = 参数 schema,Google-style docstring 能补描述

name 默认取函数名,推荐 snake_case

async def 自动支持 .ainvoke

复杂 schema 用 args_schema=PydanticBaseModel

from langchain.tools import tool

@tool
def search_docs(query: str, limit: int = 3) -> str:
"""在公司知识库里检索 query 相关条目。
Args:
query: 自然语言检索词。
limit: 最多返回条数。
"""
return docs.retriever(query, k=limit)

需 sync + async 双实现或动态构造?改用 StructuredTool.from_function(func=, coroutine=)。

create_agent — v1 主入口

返回 CompiledStateGraph(LangGraph 图),非老 AgentExecutor

system_prompt= 注意是 system_prompt,不是 prompt

支持 middleware=(PII / Summary / HITL / 自定义)

response_format=ProviderStrategy(Schema) / ToolStrategy(Schema)

checkpointer + store:memory + 长期记忆

interrupt_before / after:HITL 断点

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
model="openai:gpt-4o-mini",
tools=[search_docs],
system_prompt="你是知识库助手,简洁中文回答。",
checkpointer=InMemorySaver(),
)
# 调用
agent.invoke(
{"messages": [{"role": "user", "content": "找一下报销政策"}]},
config={"configurable": {"thread_id": "u1"}},
)

已废弃 ❌:initialize_agent / AgentExecutor / load_tools / create_react_agent(prompt=) 全部迁入 langchain-classic。新项目不要写。

Agent 执行循环与停止

model → tool_calls? → tools → model ... → AIMessage 不再调用则终止

modelLLM 推理

→
has tool_calls?条件路由

→
ToolNode并行执行

↺
→
END返回消息

自然停止:AIMessage 不含 tool_calls

兜底 LangGraph recursion_limit(默认 1000)

推荐用 ModelCallLimitMiddleware(run_limit=50) 中间件熔断

中间可加 before_model / after_model / wrap_tool_call middleware

观察工具调用:agent.stream(..., stream_mode="messages") 拿 token;stream_mode="updates" 拿工具结果。

持久化:Memory / Store

Checkpointer = 线程内短期;Store = 跨线程长期

类型实现适用

InMemorySaverlanggraph.checkpoint.memory实验 / 演示

SqliteSaverlanggraph.checkpoint.sqlite本地开发

PostgresSaverlanggraph.checkpoint.postgres生产(加密 serde 可选)

InMemoryStorelanggraph.store.memory实验 / 长记忆

PostgresStorelanggraph.store.postgres长期记忆生产

# 必传 thread_id;不传 = 不持久化
config = {"configurable": {"thread_id": "user-1"}}
agent.invoke({...}, config=config)

Checkpointer vs Store:前者存图状态、对话历史、工具中间步骤;后者存 key-value 事实/偏好跨线程共享。

🌊 流式与事件

.stream() 与 .astream_events(version="v2")

token 级别出字,看清每一步在做什么

model.stream(...):AIMessageChunk token 流

chain.stream(...):沿 LCEL 链传播最末 Runnable 的输出

astream_events(version="v2") 默认且稳定,v1 即将弃用,v3 是 beta

v2 提供 parent_ids / tags / metadata / custom events

关键事件:on_chat_model_stream / on_tool_start | end / on_retriever_start | end

async for ev in chain.astream_events(
{"question": "什么是 RAG?"}, version="v2"):
if ev["event"] == "on_chat_model_stream":
print(ev["data"]["chunk"].content, end="|")

链不流式?中间有任何先把全部结果吞下的 Runnable(parser/lambda 聚合)就会断流。

流式扩展方式

三种 stream_mode,生产常用 messages + updates 双订阅

stream_mode="updates":每步状态增量,生产首选

stream_mode="messages":LLM token 双元组(token, metadata)

stream_mode="custom":工具/节点内 get_stream_writer() 自推

多模式:stream_mode=["messages", "updates"]

for chunk in agent.stream(
{"messages": [...]},
stream_mode=["messages", "updates"],
config={"configurable": {"thread_id": "u1"}},
):
# chunk["type"] == "messages" | "updates"
...

RunnableConfig + 批处理

控制并发 / tracing / 切模型

tags、metadata、callbacks、run_name、recursion_limit

max_concurrency 控制 batch / abatch 并行上限

configurable 给 configurable_fields + LangGraph thread_id

model.batch(inputs) 默认 ThreadPool 并行,return_exceptions=True 隔离失败

chain.batch(inputs, config={
"max_concurrency": 10,
"tags": ["week2-eval"],
"callbacks": [ConsoleCallbackHandler()],
})

模型限流:改 init_chat_model(..., rate_limiter=InMemoryRateLimiter(...)) 比 max_concurrency 更精准。

📚 RAG 最小闭环

RAG 五步走(LCEL 主推写法)

load → split → embed → retrieve → generate;核心思想:并行构造 context 与 question

PyPDFLoader

→
RecursiveCharacterTextSplitter

→
OpenAIEmbeddings

→
InMemoryVectorStore

→
retriever | format_docs

→
prompt

→
model

→
StrOutputParser

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

splits = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=200).split_documents(docs)

vs = InMemoryVectorStore(embedding=OpenAIEmbeddings())
vs.add_documents(splits)
retriever = vs.as_retriever(search_kwargs={"k": 4})

prompt = ChatPromptTemplate.from_template(
"仅基于下列上下文回答,不知道就说"我不知道"。\n\n{context}\n\n问题:{question}")

def format_docs(docs): return "\n\n".join(d.page_content for d in docs)

rag_chain = (
{"context": retriever | format_docs,
"question": RunnablePassthrough()}
| prompt | ChatOpenAI(model="gpt-4o-mini")
| StrOutputParser()
)
print(rag_chain.invoke("什么是 RAG?"))

起步值:chunk_size=1000 / overlap=200(英文);中文 300-500/50-100;overlap ≈ size × 15-20%。

Retriever 三选项

默认 similarity;多样性选 mmr;门控选 score_threshold

similarity:Top-K 最近邻,可能重复

mmr:fetch_k 池中按相关 + 多样性重排(lambda_mult=0.5 常用)

similarity_score_threshold:低于阈值直接不返,可作"我不知道"兜底

metadata filter 透 search_kwargs={"filter": {...}}

# MMR
retriever = vs.as_retriever(
search_type="mmr",
search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

已 deprecated:MultiQueryRetriever / ContextualCompressionRetriever / SelfQueryRetriever 都搬到 langchain-classic;新项目用 RunnableLambda 自定义。

Vector Store 选型

起步 local;上量 PG / Pinecone / Weaviate

向量库包场景

InMemoryVectorStorelangchain-core起步、教程、测试

Chromalangchain-chroma本地持久化 + metadata filter

FAISSlangchain-community高速相似度,需手动 save/load

PGVectorlangchain-postgres已有 Postgres,免费加扩展

Pineconelangchain-pinecone托管无运维,代价是闭源

Weaviatelangchain-weaviate混合 BM25+vector,多模态

🛠️ 工程实践

重试 / 降级 / 限流

瞬态错误交给 retry,provider 切换交给 fallbacks

.with_retry(stop_after_attempt=N, wait_exponential_jitter=True)

.with_fallbacks([secondary]) 跨 provider 降级

init_chat_model(..., rate_limiter=InMemoryRateLimiter(requests_per_second=...)) 精准限流

重试只对幂等 + 瞬态错误;有副作用的工具别重试

safe = model.with_retry(
stop_after_attempt=3,
wait_exponential_jitter=True,
).with_fallbacks([backup_model])

缓存 / 结构化输出 / 调试

省 token + 把 LLM 锁到 Schema

缓存:set_llm_cache(InMemoryCache() / SQLiteCache(db_path)),相同 prompt 直接命中

结构化输出首选 model.with_structured_output(PydanticBaseModel)

结构化输出备选 model.bind_tools([schema], tool_choice="any") → AIMessage.tool_calls

本地调试:config={"callbacks": [ConsoleCallbackHandler()]}

生产追溯:LANGSMITH_TRACING=true + LANGSMITH_API_KEY

from langchain_core.globals import set_debug, set_verbose
set_debug(True) # 详细输入/输出(开发)
set_verbose(True) # 简版步骤

v0 → v1 迁移要点

"prompt= → system_prompt="、"create_react_agent → create_agent"

包拆分:langchain 主包只剩 agent helper;chains / retrievers / hub / memory 入 langchain-classic

create_react_agent → create_agent(已不在 langgraph.prebuilt)

node 名 "agent" → "model"

context 注入:config["configurable"] → 函数 context= 参数

Python 3.9 不再支持,必须 ≥ 3.10

structured output:prompted output 删除,改 ToolStrategy / ProviderStrategy

最稳读法:先看 docs.langchain.com/oss/python/migrate/langchain-v1 那张变更表。

JS/TS 对照

Python | ↔ JS .pipe() 与 RunnableSequence.from([...])

维度Python (v1)JS (v1)

核心抽象langchain-core@langchain/core

模型入口init_chat_model("openai:gpt-4o-mini")new ChatOpenAI({model: "gpt-4o-mini"}) / initChatModel(...)

管道prompt | llm | parserprompt.pipe(llm).pipe(parser)

模板ChatPromptTemplate.from_messages(...)ChatPromptTemplate.fromTemplate(...)

Agentfrom langchain.agents import create_agentimport { createAgent } from "langchain"

MemoryInMemorySaver / SqliteSaver / PostgresSaverMemorySaver / SqliteSaver / PostgresSaver

注意:JS 的 RunnableSequence.from([a, b, c]) 是 Python | 的等价;两者都需 @langchain/core 同版。

💡 速记口诀

三段式组件:model / prompt / parser,中间用 | 串起来

统一入口:init_chat_model 一行切 provider,显式 "provider:model"

结构化输出:with_structured_output > PydanticOutputParser > 拼字符串

Tool:@tool 装饰器起手;docstring = LLM 看到的说明

Agent:create_agent(model=, tools=, system_prompt=, checkpointer=);新代码不再用 initialize_agent

Memory:checkpointer = 线程内;Store = 跨线程;两者互补

thread_id:必传 {"configurable": {"thread_id": "..."}},不传 = 无持久化

流式:stream token + astream_events(version="v2") 事件;v3 beta 慎用

RAG:{context: retriever|format_docs, question: RunnablePassthrough()} 起手

chunk 起步:英文 1000/200,中文 300-500/50-100,overlap ≈ size × 15-20%

限定返回:MMR 多样性 / score_threshold 兜底"我不知道"

限流 vs 并发:rate_limiter 控制频率,max_concurrency 控制并行,各管一段

降级:.with_retry().with_fallbacks([backup]) 一气呵成

可观测:LANGSMITH_TRACING=true 一行接通 tracing;dev 用 ConsoleCallback

❌ 红线:LLMChain / initialize_agent / langchain-community / langchain.chains.* v1 起全部"够不到"

❓ 新手 FAQ

Q1: 调用 init_chat_model("claude-...") 报错 ImportError?

未装厂商包。Anthropic 必须 pip install langchain-anthropic;OpenAI 同理 langchain-openai。pip install langchain 主包不带任何模型集成。

加分:用 provider:model 前缀 + 显式 provider 双保险,避免推断漂移。

Q2: 我的 chain 不流式输出?

invoke 默认等结果全产出才返回;改用 .stream() 或 .astream_events(version="v2");链路中间不能塞"先吃完再继续"的聚合组件(如 RunnableLambda 不实现 transform)。

加分:LangChain 0.4 起 astream_events 推荐 version="v2"(v1 即将弃用)。

Q3: Agent 的 tool 被反复调用怎么办?

①检查 tool description 是否清楚;②返回 ToolMessage 是不是真让模型"看见了";③加 ToolCallLimitMiddleware(run_limit=N, exit_behavior="end") 熔断;④用 agent.stream(stream_mode="updates") 看每步,定位是哪一步贪心。

加分:prompted output + schema 太宽也是元凶——把 schema 字段收紧。

Q4: RAG 召回为空 / 答非所问?

①确认 embedding 入库和查询用同一模型;vectorstore.similarity_search(q, k=4) 直接试,跳过 retriever;MMR 失败常因 fetch_k 太小;空集合返回时改 similarity_score_threshold 或在 prompt 里明确"不知道就说我不知道"。

加分:数据变 → 索引重新建;embedding 一换 → 索引全部报废。

Q5: v0 升 v1 我应该改什么?

①装新版;from langchain.chains import ... 改 langchain_classic.chains;②from langchain.agents import create_agent;③system_prompt= 不要 prompt=;④config["configurable"] 改 context=;⑤Python ≤ 3.9 必须升 3.10+。

加分:langchainhub 已 deprecated,prompt 远端商店迁到 LangSmith Prompt Hub。

⚠️ 常见陷阱速查

陷阱现象原因解法

import langchain-community
仍能 work,但 PyPI 顶部明确 SUNSET,新版可能落不到依赖
2026-05-22 官方 sunset 决定
第三方集成改厂商专属包(langchain-openai / langchain-huggingface / langchain-postgres);loader 改 langchain-unstructured 子模块或直接换 docling / markitdown

langchain.chains.* ImportError
v1 主包已删除 langchain.chains / langchain.retrievers
v1 命名空间精简
装 langchain-classic,改 from langchain_classic.chains import ...。新项目改 LCEL

create_react_agent(prompt=...)
v1 起移出 langgraph.prebuilt 且 prompt= 改名
v1 重命名
改用 from langchain.agents import create_agent(..., system_prompt=...)

Embedding 不一致
召回命中却答非所问 / 召回率近 0
入库和检索用了不同 embedding 模型或维度
统一 EMBEDDING_MODEL 变量;换模型必重索引

chunk 过小或过大
召回命中但 LLM 拼不出来 / 大块命中后"lost in the middle"
chunk_size 与文档结构不匹配
RecursiveCharacterTextSplitter,英文 500-1000 + overlap 15-20%

同步 invoke 卡住 async
await 卡死 / 事件循环堵塞
在异步线程里跑阻塞 SDK
改 ainvoke / astream;同步阻塞放线程池

max_concurrency 设小,模型还 429
限流无效
控制并行 ≠ 控制频率
改 init_chat_model(..., rate_limiter=InMemoryRateLimiter(rps=...))

tool description 写成"占位说明"
Agent 不调用 / 调用错工具
LLM 没有"何时调用"信号
docstring 写清"何时用、参数含义、返回结构"

🚀 一键起飞:完整最小可运行

从 0 到带一个 tool + 短期记忆的 Agent

官方 Quickstart 当前写法 + 速查表常用骨干

# 1. 安装
# uv add langchain langchain-openai

# 2. 设置环境变量
# export OPENAI_API_KEY=sk-...

# 3. 跑下面这段
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

model = init_chat_model("openai:gpt-4o-mini", temperature=0)

@tool
def get_time() -> str:
"""返回服务器当前时间,精确到秒。"""
from datetime import datetime
return datetime.now().isoformat(timespec="seconds")

agent = create_agent(
model=model,
tools=[get_time],
system_prompt="你是助手,中文简洁回答。",
checkpointer=InMemorySaver(),
)

cfg = {"configurable": {"thread_id": "u1"}}
for msg in ["你好,我叫步子哥。", "现在几点了?", "我叫什么?"]:
r = agent.invoke(
{"messages": [{"role": "user", "content": msg}]},
config=cfg,
)
print(msg, "→", r["messages"][-1].content)

预期:第 3 轮能回答"步子哥",因为 thread_id="u1" 让 checkpointer 续上记忆。

LangChain 入门速查表 · 基于 LangChain v1.3+ / Python ≥ 3.10 · 2026.07 · 文白相间,以喻代说

来源: docs.langchain.com(Quickstart / Models / Agents / Tools / RAG)、reference.langchain.com、pypi.org、github.com/langchain-ai