LangChain 入门

LangChain 是构建 LLM 应用的流行框架，提供模型调用、链式处理、Agent 等能力。

为什么使用 LangChain？

直接调用 LLM API 虽然简单，但构建复杂应用时会遇到很多重复工作。LangChain 提供了一套标准化的抽象和工具。

LangChain 解决的问题：

适用场景：

核心概念

LangChain 的核心是一系列可组合的抽象：

flowchart TD
    A[LangChain] --> B[Models]
    A --> C[Prompts]
    A --> D[Chains]
    A --> E[Agents]
    A --> F[Memory]
    A --> G[Retrievers]

版本说明：

LangChain 发展迅速，API 变化较大。本文基于 LangChain 0.2+ 版本，使用最新的 LCEL 语法。

安装

# 核心包
pip install langchain langchain-core

# 模型提供商（按需安装）
pip install langchain-openai      # OpenAI
pip install langchain-anthropic   # Claude
pip install langchain-google-genai # Gemini

# 社区集成
pip install langchain-community

包结构说明：

基础使用

模型调用

LangChain 为不同模型提供统一接口：

from langchain_openai import ChatOpenAI

# 创建模型实例
llm = ChatOpenAI(model="gpt-4o-mini")

# 简单调用
response = llm.invoke("Hello")
print(response.content)

# 带消息历史
from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage(content="你是一个翻译助手"),
    HumanMessage(content="Hello")
]
response = llm.invoke(messages)

切换模型：

# OpenAI
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")

# Claude
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-sonnet-4-20250514")

# Gemini
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash")

提示词模板

模板让你可以动态生成提示词：

from langchain_core.prompts import ChatPromptTemplate

# 定义模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}"),
    ("user", "{input}")
])

# 组合成链
chain = prompt | llm

# 调用时传入变量
response = chain.invoke({"role": "翻译助手", "input": "Hello"})
print(response.content)

输出解析

将 LLM 输出转换为结构化数据：

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# 字符串输出（最常用）
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"role": "助手", "input": "你好"})
# result 是字符串

# JSON 输出
chain = prompt | llm | JsonOutputParser()
result = chain.invoke({"role": "助手", "input": "列出3种水果，用JSON格式"})
# result 是字典

LCEL (LangChain Expression Language)

LCEL 是 LangChain 的核心语法，使用 | 管道符组合组件，类似 Unix 管道。

LCEL 的优势：

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

# 定义链：Prompt → LLM → 解析器
chain = (
    ChatPromptTemplate.from_template("翻译成英文：{text}")
    | llm
    | StrOutputParser()
)

# 普通调用
result = chain.invoke({"text": "你好"})

# 流式输出
for chunk in chain.stream({"text": "你好"}):
    print(chunk, end="", flush=True)

# 批量处理
results = chain.batch([{"text": "你好"}, {"text": "世界"}])

# 异步调用
result = await chain.ainvoke({"text": "你好"})

链的组合：

# 多个链可以组合
chain1 = prompt1 | llm | StrOutputParser()
chain2 = prompt2 | llm | StrOutputParser()

# 顺序执行
combined = chain1 | chain2

# 并行执行
from langchain_core.runnables import RunnableParallel
parallel = RunnableParallel(result1=chain1, result2=chain2)

Memory (对话记忆)

Memory 让 LLM 能够”记住”之前的对话：

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

# 存储（生产环境用 Redis 等）
store = {}

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

# 带记忆的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 使用（同一 session_id 会保持对话历史）
response = chain_with_history.invoke(
    {"input": "我叫小明"},
    config={"configurable": {"session_id": "user1"}}
)

# 后续对话会记住之前的内容
response = chain_with_history.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user1"}}
)
# 输出：你叫小明

Memory 存储选择：

RAG (检索增强)

RAG 是 LangChain 最常见的应用场景，让 LLM 能够基于外部知识回答问题。

RAG 流程：

flowchart LR
    A[用户问题] --> B[检索相关文档]
    B --> C[构建 Prompt]
    C --> D[LLM 生成答案]

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. 创建向量库
embeddings = OpenAIEmbeddings()
texts = ["LangChain 是一个 LLM 应用框架", "LCEL 是 LangChain 的表达式语言"]
vectorstore = FAISS.from_texts(texts, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 2. 定义 RAG Prompt
prompt = ChatPromptTemplate.from_template("""
根据以下内容回答问题，如果内容中没有相关信息，请说"我不知道"。

参考内容：
{context}

问题：{question}

回答：""")

# 3. 构建 RAG 链
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

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

# 4. 使用
result = chain.invoke("什么是 LCEL？")

RAG 优化技巧：

Agent

Agent 是能够自主决策、调用工具的智能体。

Agent vs Chain：

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

# 定义工具（docstring 很重要，LLM 根据它决定何时调用）
@tool
def search(query: str) -> str:
    """搜索互联网获取信息。当需要查找最新信息时使用。"""
    return f"搜索结果：关于 {query} 的信息..."

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。输入应该是有效的数学表达式。"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# 创建 Agent
tools = [search, calculator]
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具来帮助用户。"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")  # 存放中间步骤
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
result = executor.invoke({"input": "计算 123 * 456 等于多少"})
print(result["output"])

Agent 执行流程：

flowchart TD
    A[用户输入] --> B{LLM 决策}
    B -->|需要工具| C[调用工具]
    C --> D[获取结果]
    D --> B
    B -->|可以回答| E[生成最终答案]

常用集成

向量数据库

# FAISS 示例
from langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_texts(texts, embeddings)

# Chroma 示例
from langchain_community.vectorstores import Chroma
vectorstore = Chroma.from_texts(texts, embeddings, persist_directory="./db")

文档加载

LangChain 支持多种文档格式：

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

# 文本文件
docs = TextLoader("file.txt").load()

# PDF（需要 pip install pypdf）
docs = PyPDFLoader("file.pdf").load()

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

# CSV
docs = CSVLoader("data.csv").load()

文本分割

长文档需要分割成小块才能有效检索：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每块最大字符数
    chunk_overlap=200,    # 块之间重叠字符数
    separators=["\n\n", "\n", "。", " ", ""]  # 分割优先级
)
chunks = splitter.split_documents(docs)

分割策略：

最佳实践

1. 使用 LCEL

LCEL 是 LangChain 的现代写法，旧的 Chain 类已弃用：

# ✅ 推荐：LCEL 写法
chain = prompt | llm | parser

# ❌ 旧写法（已弃用）
chain = LLMChain(llm=llm, prompt=prompt)

2. 流式优先

流式输出能显著提升用户体验：

# 同步流式
for chunk in chain.stream(input):
    print(chunk, end="", flush=True)

# 异步流式（推荐）
async for chunk in chain.astream(input):
    yield chunk

3. 错误处理

from langchain_core.runnables import RunnableConfig

# 配置重试和超时
result = chain.invoke(
    input,
    config=RunnableConfig(
        max_retries=3,
        timeout=30
    )
)

# 带回退的链
from langchain_core.runnables import RunnableLambda

fallback_chain = chain.with_fallbacks([backup_chain])

4. 调试技巧

# 开启详细日志
import langchain
langchain.debug = True

# 或使用 verbose
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

常见问题

Q: LangChain vs 直接调用 API？

Q: LangChain vs LlamaIndex？

Q: 版本兼容问题？

LangChain 更新频繁，建议：

• 锁定版本号
• 关注官方迁移指南
• 使用 langchain-core 中的稳定 API

相关文档

• 13-大模型-Agent与工具调用
• 15-大模型-Embedding与向量检索
• 20-大模型-知识库问答系统
• 08-大模型-SDK使用指南