LangChain 完全指南

官方文档：https://python.langchain.com/docs/introduction/
适用版本：langchain 0.3.x / langchain-core 0.3.x（2026-05-07 核实）

概述

LangChain 是构建 LLM 应用的开源框架，核心价值在于：提供统一抽象层连接各家 LLM API，并通过 LCEL（LangChain Expression Language）将提示词、模型、解析器、记忆、检索器等组件像搭积木一样组合成链（Chain）。

对比主要替代方案：

• 直接调用 OpenAI SDK：简单任务够用，但多步骤 RAG / Agent 场景需要大量胶水代码
• LlamaIndex：专注于 RAG 和结构化数据检索，LangChain 则更通用，覆盖 Agent / Workflow
• LangGraph：LangChain 团队推出的有向图工作流框架，适合复杂多步骤 Agent；LangChain 0.3 以后推荐用 LangGraph 替代复杂 Chain

适合场景：RAG（检索增强生成）应用、多步骤 Agent、LLM 应用原型快速验证、需要接入多种模型提供商并保持可替换性。不适合场景：只需一次 LLM 调用的简单场景（直接用 SDK 更轻量）、需要精细状态管理的复杂 Agent（用 LangGraph）。

安装

pip install langchain langchain-openai langchain-anthropic   # 核心 + 常用提供商
pip install langchain-community                              # 社区集成（向量库、文档加载器等）
pip install langchain-chroma faiss-cpu                       # 向量数据库
pip install pypdf python-docx unstructured                   # 文档加载器依赖

LangChain 0.3 以后拆分为多个包：

核心概念：Runnable 与 LCEL

LCEL（LangChain Expression Language）是 LangChain 0.2+ 的核心编程模型。所有组件实现 Runnable 协议，可以用 | 管道运算符组合成链。

Runnable 协议

所有 LCEL 组件都实现以下方法：

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

# LCEL 管道：prompt | model | parser
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{question}"),
])
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | model | parser

# 同步调用
result = chain.invoke({"question": "What is RAG?"})

# 流式输出
for chunk in chain.stream({"question": "Explain LLMs"}):
    print(chunk, end="", flush=True)

# 异步调用
import asyncio
result = asyncio.run(chain.ainvoke({"question": "Hello"}))

# 批量调用
results = chain.batch([
    {"question": "What is Python?"},
    {"question": "What is TypeScript?"},
])

ChatModel — 语言模型接口

ChatOpenAI

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,
    max_tokens=1024,
    timeout=30,
    max_retries=2,
    api_key="sk-...",         # 默认读取 OPENAI_API_KEY 环境变量
    base_url=None,            # 自定义 API endpoint（用于代理或兼容 API）
    streaming=False,
)

ChatAnthropic（Claude）

from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(
    model="claude-opus-4-7",
    temperature=0,
    max_tokens=4096,
    api_key="...",            # 默认读取 ANTHROPIC_API_KEY
)

直接调用模型（消息格式）

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

messages = [
    SystemMessage(content="You are a Python expert."),
    HumanMessage(content="What is a list comprehension?"),
]

response = llm.invoke(messages)
print(response.content)
print(response.usage_metadata)   # token 用量

PromptTemplate — 提示词模板

ChatPromptTemplate

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# from_messages：最常用的构造方式
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a {role} assistant. Answer in {language}."),
    MessagesPlaceholder("history"),   # 注入对话历史列表
    ("human", "{input}"),
])

# 部分填充（只填一部分变量）
partial_prompt = prompt.partial(language="Chinese")

# 格式化（生成消息列表）
messages = partial_prompt.invoke({
    "role": "customer service",
    "history": [],
    "input": "How do I reset my password?",
})

MessagesPlaceholder

from langchain_core.prompts import MessagesPlaceholder

# 在 prompt 中插入动态消息列表（用于多轮对话历史）
MessagesPlaceholder(variable_name="history", optional=False)

PromptTemplate（非对话场景）

from langchain_core.prompts import PromptTemplate

template = PromptTemplate.from_template(
    "Summarize the following text in {num_words} words:\n\n{text}"
)
# 等价于
template = PromptTemplate(
    template="Summarize the following text in {num_words} words:\n\n{text}",
    input_variables=["num_words", "text"],
)

OutputParser — 输出解析器

StrOutputParser

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
chain = prompt | llm | parser
result = chain.invoke({"input": "hello"})  # str，不是 AIMessage 对象

JsonOutputParser

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class Movie(BaseModel):
    title: str = Field(description="电影标题")
    year: int = Field(description="发行年份")
    rating: float = Field(description="评分 0-10")

parser = JsonOutputParser(pydantic_object=Movie)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract movie info. {format_instructions}"),
    ("human", "{input}"),
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | llm | parser
result = chain.invoke({"input": "Inception (2010) is a masterpiece, 9.3/10"})
# Movie(title='Inception', year=2010, rating=9.3)

Memory — 对话记忆

LangChain 0.3 推荐在代码层面手动管理对话历史，而非使用已废弃的 ConversationBufferMemory 等类。

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

# 创建 chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder("history"),
    ("human", "{input}"),
])
chain = prompt | llm | StrOutputParser()

# 每个 session 有独立的历史存储
store = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 包装 chain，自动注入历史
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# 多轮对话
config = {"configurable": {"session_id": "user-123"}}
chain_with_history.invoke({"input": "My name is Alice."}, config=config)
chain_with_history.invoke({"input": "What's my name?"}, config=config)
# "Your name is Alice."

RAG — 检索增强生成

RAG 是 LangChain 最典型的应用场景，流程为：加载文档 → 分块 → 嵌入 → 存入向量库 → 检索 → 注入 LLM。

文档加载器

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

# PDF
loader = PyPDFLoader("document.pdf")
docs = loader.load()  # list[Document]

# 网页
loader = WebBaseLoader("https://example.com/article")
docs = loader.load()

# 目录批量加载
loader = DirectoryLoader("./docs/", glob="**/*.md")
docs = loader.load()

文本分块

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块最大字符数
    chunk_overlap=200,    # 相邻块重叠字符数（保留上下文）
    separators=["\n\n", "\n", "。", ".", " ", ""],
)

chunks = splitter.split_documents(docs)

向量存储

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 从文档列表创建向量库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",   # 持久化到磁盘
)

# 加载已有向量库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
)

# 相似度检索
retriever = vectorstore.as_retriever(
    search_type="similarity",     # 或 "mmr"（最大边际相关性，减少重复）
    search_kwargs={"k": 4},       # 返回 top-4 结果
)

results = retriever.invoke("What is RAG?")

完整 RAG Chain

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

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

prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer the question based only on the following context:

{context}

If the answer is not in the context, say "I don't know"."""),
    ("human", "{question}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("What is the main benefit of RAG?")

Agent — 工具调用代理

Agent 让 LLM 自主决定调用哪些工具来完成任务。LangChain 0.3+ 推荐用 create_tool_calling_agent 配合 AgentExecutor，或迁移至 LangGraph。

from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_openai import ChatOpenAI

@tool
def search_web(query: str) -> str:
    """Search the web for current information about a topic."""
    # 实际项目中接入 Tavily / SerpAPI
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression. Input should be a valid Python math expression."""
    return eval(expression)  # 生产环境用 numexpr 或安全沙箱

tools = [search_web, calculate]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Use tools to answer questions."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),   # Agent 中间推理步骤
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,          # 打印每步推理
    max_iterations=10,     # 最多迭代次数，防止死循环
    handle_parsing_errors=True,
)

result = executor.invoke({"input": "What is 123 * 456, and who invented Python?"})
print(result["output"])

@tool 装饰器

完整示例：带记忆的 RAG 问答系统

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

# 1. 加载并索引文档
loader = PyPDFLoader("knowledge_base.pdf")
docs = loader.load()

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

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 2. 构建 RAG Chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", """Given chat history and the latest user question,
formulate a standalone question that can be understood without the chat history.
Do NOT answer the question, just reformulate it."""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer the question based on the context:\n\n{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

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

# 问题改写 chain（结合历史将模糊问题改写为独立问题）
contextualize_chain = contextualize_prompt | llm | StrOutputParser()

def get_retrieval_chain(input_dict):
    question = input_dict["input"]
    history = input_dict.get("chat_history", [])
    if history:
        question = contextualize_chain.invoke({"input": question, "chat_history": history})
    return retriever.invoke(question)

rag_chain = (
    RunnablePassthrough.assign(context=get_retrieval_chain | format_docs)
    | qa_prompt
    | llm
    | StrOutputParser()
)

# 3. 添加记忆
store = {}
def get_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

chain_with_memory = RunnableWithMessageHistory(
    rag_chain,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# 4. 多轮对话
cfg = {"configurable": {"session_id": "demo"}}
print(chain_with_memory.invoke({"input": "What is the main topic of the document?"}, cfg))
print(chain_with_memory.invoke({"input": "Can you elaborate on that?"}, cfg))

最佳实践

用 LCEL 替代旧版 Chain 类：LLMChain、ConversationalRetrievalChain 等在 0.3 中已软废弃，LCEL 的 | 组合更透明、可调试、可流式，且能直接用 RunnablePassthrough.assign() 注入中间变量。

# 旧式（废弃中）
from langchain.chains import LLMChain
chain = LLMChain(llm=llm, prompt=prompt)

# 现代 LCEL
chain = prompt | llm | StrOutputParser()

将 API Key 放入环境变量，不硬编码：ChatOpenAI() 默认读取 OPENAI_API_KEY 环境变量，无需在代码中传入。

# 正确：.env 文件 + python-dotenv
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini")  # 自动读取环境变量

# 错误：硬编码密钥
llm = ChatOpenAI(api_key="sk-proj-abc123...")

RAG 中使用 chunk_overlap 保留上下文边界：当知识库文档被分割时，关键信息往往出现在块的边界处。chunk_overlap=200 确保相邻块共享 200 个字符，减少信息截断导致的答案错误。

使用 LangSmith 追踪和调试：LangChain 所有调用都可以接入 LangSmith，自动记录每一步的输入输出、token 用量、延迟，是调试复杂 Chain / Agent 的必备工具。

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."
# 之后所有 chain.invoke() 都会自动上报到 LangSmith

RAG 检索用 MMR 减少重复内容：默认相似度检索可能返回多个高度相似的块（来自同一段落的相邻分割），用最大边际相关性（MMR）检索在保持相关性的同时增加多样性。

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

为 Agent 的工具提供清晰的 docstring：LLM 完全依赖工具的名称和 docstring 来决定何时调用、传什么参数。模糊的描述直接导致工具调用错误率上升。

# 清晰描述工具的用途、输入格式、返回值
@tool
def get_stock_price(ticker: str) -> str:
    """Get the current stock price for a given ticker symbol.
    
    Args:
        ticker: Stock ticker symbol, e.g. 'AAPL', 'GOOGL', 'MSFT'
    
    Returns:
        Current price as a string, e.g. '$182.50'
    """
    ...

常见陷阱

陷阱：RAG 答案引用幻觉上下文

现象： 模型给出看似合理但实际与检索结果无关的答案，像是"自己编的"。

原因： Prompt 中没有明确限制模型只能基于提供的 context 回答，模型会结合训练数据补充知识。

解决： 在 system prompt 中明确约束，并让模型在无法从上下文回答时声明。

system_prompt = """Answer ONLY based on the following context. 
If the answer cannot be found in the context, respond with:
"I don't have enough information in the provided documents to answer this question."

Context:
{context}"""

陷阱：Agent 陷入工具调用死循环

现象： Agent 反复调用同一工具，达到 max_iterations 上限后才停止，输出无意义结果。

原因： 工具返回值格式不符合 LLM 预期，LLM 不断重试；或工具 docstring 描述了错误的返回格式。

解决： 设置合理的 max_iterations（默认 15 通常太高）；工具返回值加上结构化的成功 / 失败标识；确保 docstring 描述与实际返回一致。

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,             # 限制迭代次数
    max_execution_time=30.0,      # 30 秒超时
    handle_parsing_errors=True,   # 解析错误时自动恢复
)

陷阱：流式输出与 RunnableWithMessageHistory 不兼容

现象： 在带记忆的 chain 上调用 .stream() 时，历史消息不被正确保存（只保存了第一个 chunk）。

原因： RunnableWithMessageHistory 在 stream 模式下需要收集完所有 chunks 才能将 AI 回复写入历史，部分旧版本实现有 bug。

解决： 升级到 langchain-core >= 0.3；流式输出时手动收集完整回复后再写入历史，或使用 LangGraph 的 streaming 支持。

# 手动流式收集并更新历史
full_response = ""
for chunk in chain.stream({"input": question, "chat_history": history}):
    full_response += chunk
    print(chunk, end="", flush=True)

# 手动追加到历史
history.add_user_message(question)
history.add_ai_message(full_response)

陷阱：向量库冷启动每次重新 embed

现象： 应用重启后向量库数据丢失，需要重新加载文档并 embed，每次启动耗时几分钟且产生额外 API 费用。

原因： Chroma.from_documents() 每次都创建新的内存库；未设置 persist_directory 则不会持久化。

解决： 区分"首次建库"和"加载已有库"的逻辑。

import os

persist_dir = "./chroma_db"

if os.path.exists(persist_dir) and os.listdir(persist_dir):
    # 加载已有向量库，不重新 embed
    vectorstore = Chroma(persist_directory=persist_dir, embedding_function=embeddings)
else:
    # 首次建库
    vectorstore = Chroma.from_documents(
        chunks, embeddings, persist_directory=persist_dir
    )

参见

• HuggingFace Transformers完全指南
• scikit-learn完全指南
• FastAPI完全指南
• Redis完全指南