---

关于作者

• 深耕领域：大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
• 技术栈：Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
• 工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案

「让 AI 交互更智能，让技术落地更高效」
欢迎技术探讨与项目合作，解锁大模型与智能交互的无限可能！

---

在大模型应用开发中，你是否曾被这些问题困扰：

• 如何让大模型调用外部工具（如数据库、API）？
• 如何将大模型与自有知识库结合，实现精准问答？
• 如何构建多轮对话系统，让模型记住上下文？
• 如何将大模型能力封装为可复用的组件？

LangChain 作为大模型应用开发的 “瑞士军刀”，通过一系列 模块化核心组件 ，让开发者可以像搭乐高积木一样快速构建复杂的大模型应用。本文将从底层原理出发，带你彻底搞懂 LangChain 的核心组件，以及如何用它们构建生产级大模型应用。

---

一、LangChain 的核心设计理念：模块化与可组合性

LangChain 的核心理念是 将大模型应用拆解为独立组件，通过组合实现复杂功能 。这些组件可以独立使用，也可以灵活组合，覆盖了大模型应用开发的全流程：

每个组件都有明确的职责，开发者可以根据需求选择合适的组件进行组合，大大降低了大模型应用的开发门槛。

---

二、核心组件 1：模型层（Models）—— 大模型的 “接口层”

提示：运行以下示例前，请确保已安装最新版核心包及插件：

pip install -U langchain langchain-openai langchain-community langchain-text-splitters faiss-cpu

模型层是 LangChain 与大模型交互的核心组件。在 LangChain 1.0+ 中，所有模型都继承自 Runnable 接口，支持统一的 invoke 调用。

1. 核心类型

（1）LLM（大语言模型）

处理纯文本输入输出。

from langchain_openai import OpenAI

# 初始化OpenAI LLM (需要环境变量 OPENAI_API_KEY)
# llm = OpenAI(model_name="gpt-3.5-turbo-instruct", temperature=0.7)

# 使用 Mock LLM 演示（无需 API Key）
from langchain_core.language_models.llms import LLM
from typing import Optional, List, Any
class MockLLM(LLM):
    def _call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs: Any) -> str:
        return f"这是关于 {prompt[:10]} 的技术博客内容..."
    @property
    def _llm_type(self) -> str: return "mock"

llm = MockLLM()

# 推荐使用 invoke 方法
response = llm.invoke("请写一篇关于LangChain的技术博客")
print(response)

（2）ChatModel（对话模型）

专门处理对话场景，输入输出均为消息对象（Messages）。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# chat_model = ChatOpenAI(model="gpt-3.5-turbo")

# 使用 Mock ChatModel 演示
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.outputs import ChatResult, ChatGeneration
from langchain_core.messages import AIMessage
class MockChatModel(BaseChatModel):
    def _generate(self, messages, stop=None, **kwargs):
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content="你好！我是 AI 技术博主。"))])
    @property
    def _llm_type(self): return "mock"

chat_model = MockChatModel()

# 多轮对话
messages = [
    SystemMessage(content="你是一名专业的技术博客作者"),
    HumanMessage(content="请写一篇关于LangChain核心组件的技术博客")
]
response = chat_model.invoke(messages)
print(response.content)

（3）EmbeddingModel（嵌入模型）

将文本转换为向量，用于语义检索。

from langchain_openai import OpenAIEmbeddings

# embeddings = OpenAIEmbeddings()

# 使用 Mock Embeddings 演示
from langchain_core.embeddings import Embeddings
class MockEmbeddings(Embeddings):
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [[0.1] * 1536 for _ in texts]
    def embed_query(self, text: str) -> List[float]:
        return [0.1] * 1536

embeddings = MockEmbeddings()

# 生成文本向量
text = "LangChain is a framework for building LLM applications"
vector = embeddings.embed_query(text)
print(f"向量维度：{len(vector)}")

2. 关键特性

• 统一接口 ：不同大模型的接口差异被抽象为统一的 API，开发者无需为每个模型单独适配；
• 参数配置 ：支持温度（temperature）、最大 Token 数（max_tokens）等参数的全局配置；
• 异步支持 ：所有模型都支持异步调用，提升大模型应用的并发性能；
• 模型缓存 ：支持缓存模型响应，避免重复调用相同请求，降低成本。

---

三、核心组件 2：数据连接层（Data Connection）—— 大模型的 “知识库”

数据连接层负责将外部数据（如文档、数据库、API 数据）与大模型结合，实现 检索增强生成（RAG）。

1. 核心组件

（1）Document Loaders（文档加载器）

将外部文档加载为 LangChain 的 Document 对象。目前这些加载器主要位于 langchain-community 包中。

from langchain_community.document_loaders import PyPDFLoader

# 加载PDF文档
# loader = PyPDFLoader("langchain_whitepaper.pdf")
# documents = loader.load()
# print(f"加载的文档页数：{len(documents)}")

（2）Text Splitters（文本分割器）

将长文档分割为小片段。LangChain 现在拥有独立的 langchain-text-splitters 包。

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 初始化文本分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 每个片段的最大字符数
    chunk_overlap=200  # 片段之间的重叠部分
)

# 模拟分割文档
# split_documents = text_splitter.split_documents(documents)

（3）Vector Stores（向量数据库）与 Retrievers（检索器）

存储向量并支持语义检索。建议使用 langchain-chroma 或 FAISS。

from langchain_community.vectorstores import FAISS

# 模拟文档数据
from langchain_core.documents import Document
docs = [Document(page_content="LangChain 是一个大模型应用开发框架")]

# 使用前面定义的 embeddings (MockEmbeddings) 创建本地向量库
vector_store = FAISS.from_documents(docs, embeddings)

# 创建检索器
retriever = vector_store.as_retriever(search_kwargs={"k": 1})

# 执行检索
retrieved_docs = retriever.invoke("什么是 LangChain？")
print(f"检索结果：{retrieved_docs[0].page_content}")

---

四、核心组件 3：链（Chains）与 LCEL —— 组件的 “流水线”

链（Chains）负责将多个组件组合成一个可执行的流水线。在 LangChain 1.0+ 中，推荐使用 LangChain Expression Language (LCEL) 来构建链，它比传统的 Chain 类更灵活、更易于调试。

1. 现代组合方式：LCEL

LCEL 允许你使用 | 运算符将 Prompt、模型和输出解析器连接起来。

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

# 创建 Prompt 模板
prompt = ChatPromptTemplate.from_template("请写一篇关于 {topic} 的技术博客")

# 构建 LCEL 链 (Prompt -> LLM -> OutputParser)
chain = prompt | chat_model | StrOutputParser()

# 执行链
response = chain.invoke({"topic": "LangChain 1.0 架构"})
print(response)

2. 传统链（Legacy Chains）

虽然官方推荐 LCEL，但传统的 LLMChain、RetrievalQA 等在旧代码库中依然常见。

# 旧版导入方式
try:
    from langchain.chains import LLMChain
except ImportError:
    print("请安装 langchain 包以使用旧版链")

3. 为什么选择 LCEL？

LCEL 不仅仅是一种语法糖，它为现代大模型应用提供了核心工程能力：

• 流式处理 (Streaming)：自动支持从模型到输出解析器的流式输出；
• 并行执行 (Parallelism)：使用 RunnableParallel 轻松实现并发调用；
• 异步支持 (Async)：所有 LCEL 链原生支持 ainvoke；
• 中间步骤可视化：可以轻松追踪链中的每一个环节。

---

五、核心组件 4：记忆（Memory）—— 大模型的 “对话上下文”

记忆组件负责存储对话历史。在 1.0+ 架构中，推荐直接在 LCEL 链中管理消息历史，或者使用 RunnableWithMessageHistory。

1. 基础记忆：ConversationBufferMemory

这是最简单的记忆组件，存储完整的对话历史。

try:
    from langchain.memory import ConversationBufferMemory
except ImportError:
    print("请安装 langchain 包以使用旧版记忆组件")

# 初始化记忆组件
memory = ConversationBufferMemory()

# 在传统对话链中使用
# from langchain.chains import ConversationChain
# conversation = ConversationChain(llm=llm, memory=memory)

2. 现代记忆管理方式

现代做法是使用 ChatMemoryHistory 配合 RunnableWithMessageHistory，这样可以更好地控制状态持久化。

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

# 初始化消息历史
history = ChatMessageHistory()

# 现代链配合历史记录（伪代码示例）
# with_history = RunnableWithMessageHistory(chain, lambda session_id: history)

---

六、核心组件 5：代理（Agents）—— 大模型的 “决策中心”

代理（Agents）让大模型可以根据用户请求，自主决策调用哪些工具。

1. 现代 Agent 构建方式

LangChain 现在推荐使用 create_react_agent 或更强大的 LangGraph。以下是一个标准的 ReAct Agent 构建流程：

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

# 1. 定义工具集
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
    """获取指定城市的实时天气"""
    return f"{city} 的天气是：晴，25度"

tools = [get_weather]

# 2. 获取 ReAct 提示词模板 (需要联网或手动定义)
# prompt = hub.pull("hwchase17/react")
# 这里我们模拟一个简单的 Prompt 模板
from langchain_core.prompts import PromptTemplate
template = "尽力回答以下问题。你可以使用以下工具：{tools}... 问题：{input} {agent_scratchpad}"
prompt = PromptTemplate.from_template(template)

# 3. 构建 Agent 运行器
# agent = create_react_agent(chat_model, tools, prompt)
# executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 4. 执行任务
# response = executor.invoke({"input": "北京今天天气怎么样？"})

注意：对于更复杂的逻辑控制，推荐学习 LangGraph，它是目前构建生产级 AI Agent 的首选框架。

---

七、核心组件 6：工具（Tools）—— 大模型的 “手脚”

工具是大模型与外部系统交互的接口。在 1.0+ 中，推荐使用 @tool 装饰器。

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

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")

@tool("search_tool", args_schema=SearchInput)
def search(query: str) -> str:
    """在互联网上搜索信息"""
    return f"搜索结果: {query} 的相关信息..."

---

八、组件组合：构建生产级大模型应用 (RAG 实战)

通过组合 LangChain 的核心组件，我们可以快速构建复杂的生产级大模型应用。以下是一个典型的 检索增强生成 (RAG) 系统示例（使用 Mock 组件演示可运行流程）：

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

# 1. 模拟检索器 (Retriever)
class MockRetriever:
    def invoke(self, query: str):
        return "LangChain 的核心价值在于其模块化设计和强大的生态系统。"

retriever = MockRetriever()

# 2. 定义 Prompt 模板
template = """基于以下内容回答问题:
{context}

问题: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 3. 构建 LCEL RAG 链
# 使用前面定义的 chat_model (MockChatModel)
rag_chain = (
    {"context": retriever.invoke, "question": RunnablePassthrough()}
    | prompt
    | chat_model
    | StrOutputParser()
)

# 4. 执行问答
response = rag_chain.invoke("LangChain 的核心价值是什么？")
print(f"RAG 回答: {response}")

---

九、总结：LangChain 的核心价值

LangChain 1.0+ 的核心价值在于 标准化与工程化。

• 模型层：通过 Runnable 接口统一了所有组件的调用方式；
• 数据连接层：标准化的 RAG 流程，极大降低了知识库构建难度；
• LCEL：声明式的链式表达，让复杂的逻辑变得清晰且易于并行化；
• 生态化：丰富的社区插件（Community）和合作伙伴集成（OpenAI, Anthropic 等）。

如果你想构建稳定、可扩展的大模型应用，掌握 LangChain 1.0+ 的模块化设计和 LCEL 是必经之路！