作者：CSDN博客
LangChain是一个强大的开源LLM应用开发框架，本文深入解析了其四大核心概念：Model I/O（与LLM交互）、Chain（串联组件的工作流）、Memory（对话状态管理）和LCEL（声明式组件组合语法）。文章通过实例展示了如何使用这些概念构建自动化工作流，并介绍了2026年的新特性及进阶学习路径，帮助读者全面掌握LangChain，轻松入门大模型开发。
一、引言
LangChain 是由 Harrison Chase 于2022年10月发起的开源LLM应用开发框架，比ChatGPT问世还要早一个月。经过近三年的发展，LangChain已从单一开源包成长为覆盖开发、调试、部署全流程的完整生态系统。
截至2026年，LangChain在GitHub上已获得超过90,000 Stars，每月数百万次下载，成为企业级LLM应用开发的标准基础设施。本文将深入解析LangChain的四大核心概念：Model I/O、Chain、Memory、LCEL，帮助你建立完整的知识体系。
二、整体架构概览
2.1 分层架构设计
LangChain采用分层架构，不同层级各司其职又可协同工作：
LangChain生态
应用层langgraphAgent编排、状态管理、工作流
组件层langchainChains、Agents、Retrieval
集成层langchain-community第三方模型、工具、向量库
核心层langchain-core基础抽象、LCEL、Runnable接口
• langchain-core：提供基础抽象与LCEL，是组件协同的核心• langchain-community：第三方集成模块，覆盖Model I/O、Retrieval、Tool等• langchain：包含Chains、Agents、Retrieval等核心业务组件• langgraph：编排多个节点，负责整个工作流的调度与状态跳转
2.2 核心组件关系
用户输入
Prompt Templates
LLM
Output Parsers
最终输出
Memory
Chain
Tools
三、Model I/O：与LLM的沟通桥梁
Model I/O是应用与LLM交互的核心模块，类似于JDBC与数据库的关系。其核心价值在于解耦应用逻辑与底层模型实现，让你可以自由切换不同提供商（OpenAI、Anthropic、Google等）而不改变业务代码。
3.1 三步工作流程
Model I/O的工作流程分为三个关键步骤：
Parse 输出解析
Predict 模型调用
Format 输入格式化
Prompt Templates
Chat Models
Output Parsers
3.2 Prompt Templates
Prompt Templates 用于结构化提示词，支持变量替换和复用：- from langchain_core.prompts import ChatPromptTemplate# 方式一：from_messages（推荐）prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个{role}，专门帮助用户解决{topic}问题"), ("human", "我的问题是：{question}")])# 方式二：from_template（传统方式）prompt = ChatPromptTemplate.from_template( "你是{role}，请用{style}风格回答以下问题：{question}")# 格式化输入formatted_prompt = prompt.format( role="技术顾问", topic="编程", question="Python如何处理异常？", style="简洁专业")print(formatted_prompt)
- System: 你是一个技术顾问，专门帮助用户解决编程问题Human: 我的问题是：Python如何处理异常？
LangChain提供统一的模型接口，支持多种提供商：- from langchain_openai import ChatOpenAIfrom langchain_anthropic import ChatAnthropicfrom langchain_google_genai import ChatGoogleGenerativeAI# OpenAIopenai_llm = ChatOpenAI( model="gpt-4o", temperature=0.7, streaming=True # 启用流式输出)# Anthropicclaude_llm = ChatAnthropic( model="claude-3-5-sonnet-20241022", temperature=0.7)# Google Geminigemini_llm = ChatGoogleGenerativeAI( model="gemini-2.0-flash-exp", temperature=0.7)
Output Parsers 将原始输出转换为结构化数据：- from langchain_core.output_parsers import StrOutputParserfrom langchain_core.output_parsers import JsonOutputParserfrom langchain.output_parsers import PydanticOutputParserfrom pydantic import BaseModel# 字符串解析器（最常用）str_parser = StrOutputParser()# JSON解析器json_parser = JsonOutputParser()# Pydantic解析器（推荐用于结构化输出）class Answer(BaseModel): result: str confidence: float sources: list[str]pydantic_parser = PydanticOutputParser(pydantic_object=Answer)# 使用示例response = llm.invoke("What is 2+2?")parsed = str_parser.invoke(response)print(parsed) # "2"
将三个组件串联成完整流程：- from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParser# 初始化模型llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 创建Prompt模板prompt = ChatPromptTemplate.from_template( "用一句话解释{concept}，用中文回答")# 创建输出解析器parser = StrOutputParser()# 方式一：传统方式（手动调用）formatted_prompt = prompt.format(concept="LLM")response = llm.invoke(formatted_prompt)result = parser.invoke(response)print(result)# 方式二：LCEL管道方式（推荐）chain = prompt | llm | parserresult = chain.invoke({"concept": "LLM"})print(result)
四、Chains：串联组件的工作流
Chains（链）用于将多个组件串联成自动化工作流，实现步骤化任务处理。
4.1 LLMChain
最基本的链，用于将Prompt模板、LLM、输出解析器组合：- from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain.chains import LLMChainllm = ChatOpenAI(model="gpt-4o-mini")prompt = ChatPromptTemplate.from_template( "帮我写一首关于{topic}的诗")chain = LLMChain(llm=llm, prompt=prompt)result = chain.run(topic="春天")print(result)
顺序执行多个链：- from langchain.chains import SequentialChainfrom langchain.chains import LLMChainfrom langchain_core.prompts import ChatPromptTemplate# 链1：生成标题title_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("为以下内容生成一个标题：{content}"), output_key="title")# 链2：生成摘要summary_chain = LLMChain( llm=llm, prompt=ChatPromptTemplate.from_template("用50字概括以下内容：{content}"), output_key="summary")# 组合顺序链sequential_chain = SequentialChain( chains=[title_chain, summary_chain], input_variables=["content"], output_variables=["title", "summary"])result = sequential_chain.invoke({"content": "Python是一门易学难精的编程语言..."})print(result)
| 特性 | 传统方式 (LLMChain) | LCEL方式 | | 语法 | 面向对象，方法调用 | 管道运算符 ` | | 灵活性 | 较低，固定模式 | 高，可自由组合 | | 并行支持 | 需额外处理 | 原生支持 | | 流式输出 | 需额外配置 | 原生支持 | | 异步支持 | 需异步封装 | 原生async/await |
五、LCEL语法：LangChain Expression Language
LCEL是LangChain最强大的特性之一，它将所有组件统一为Runnable接口，通过管道运算符实现声明式组合。
5.1 Runnable接口
LangChain v1.x中，所有组件都实现了Runnable接口：- from langchain_core.runnables import Runnable# Runnable接口的核心方法class Runnable: def invoke(self, input, config=None): """同步调用""" pass async def ainvoke(self, input, config=None): """异步调用""" pass def stream(self, input, config=None): """流式输出""" pass def batch(self, inputs, config=None): """批量处理""" pass
| 运算符将前一个组件的输出作为下一个组件的输入：- # 基本语法chain = component_a | component_b | component_c# 等价于result = component_c.invoke(component_b.invoke(component_a.invoke(input)))
传统方式：- from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserllm = ChatOpenAI(model="gpt-4o-mini")prompt = ChatPromptTemplate.from_template("用中文介绍{topic}")parser = StrOutputParser()# 手动串联formatted = prompt.format(topic="人工智能")response = llm.invoke(formatted)result = parser.invoke(response)print(result)
- # 一行搞定chain = prompt | llm | parserresult = chain.invoke({"topic": "人工智能"})print(result)
LCEL支持并行处理和条件分支：- from langchain_core.runnables import RunnableParallel, RunnableBranch# 并行执行多个分支parallel_chain = RunnableParallel( chinese = prompt | llm | parser, english = (prompt | llm | parser).bind(language="en"), code = code_prompt | llm | parser)# 条件分支branch_chain = RunnableBranch( (lambda x: x.get("type") == "code", code_chain), (lambda x: x.get("type") == "data", data_chain), default_chain # 默认链)
LCEL原生支持流式输出：- chain = prompt | llm | parser# 流式输出（一个字一个字显示）for chunk in chain.stream({"topic": "量子计算"}): print(chunk, end="", flush=True)
| 优势 | 说明 | | 简洁性 | 用管道运算符替代嵌套函数调用 | | 可组合性 | 组件可自由组合，易于扩展 | | 原生支持 | 并行、流式、异步开箱即用 | | 可观测性 | 易于调试和监控 | | 性能 | 官方称并行处理效率提升30%-50% |
六、Memory：对话状态的守护者
Memory（内存）用于在对话过程中维护状态，解决LLM上下文窗口有限的问题。
6.1 短期记忆 vs 长期记忆
Memory 分类
长期记忆跨会话持久化·知识积累
VectorStore
RetrieverMemory
短期记忆单次会话·上下文保持
BufferMemory
BufferWindowMemory
SummaryMemory
6.2 ConversationBufferMemory
最简单的内存，保存完整对话历史：- from langchain.memory import ConversationBufferMemoryfrom langchain.chains import LLMChainfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplatellm = ChatOpenAI(model="gpt-4o-mini")# 创建带记忆的链memory = ConversationBufferMemory( memory_key="history", return_messages=True # 返回消息对象列表)prompt = ChatPromptTemplate.from_template( """基于以下对话历史回答问题：历史：{history}问题：{question}""")chain = LLMChain(llm=llm, prompt=prompt, memory=memory)# 多轮对话print(chain.invoke({"question": "我叫张三"}))print(chain.invoke({"question": "我叫什么名字？"}))
限制保存最近N轮对话，避免历史过长：- from langchain.memory import ConversationBufferWindowMemorymemory = ConversationBufferWindowMemory( k=3, # 只保留最近3轮对话 memory_key="chat_history", return_messages=True)
自动总结对话要点，适合长对话：- from langchain.memory import ConversationSummaryMemorymemory = ConversationSummaryMemory( llm=llm, # 用于生成摘要的LLM memory_key="summary", return_messages=True)
VectorStoreRetrieverMemory 是LangChain提供的长期记忆组件，它将对话内容存储在向量数据库中，通过语义相似性检索来回忆相关信息。与短期记忆不同，它不显式跟踪对话顺序，而是根据语义相关性动态检索最"显著"的记忆片段。
6.5.1 核心特点
| 特性 | 说明 | | 存储方式 | 向量数据库（FAISS、Chroma、Pinecone等） | | 检索方式 | 语义相似性搜索，而非时间顺序 | | 适用场景 | 跨会话记忆、语义检索、长周期知识积累 | | 优势 | 支持海量记忆、按语义检索、不受token限制 |
6.5.2 工作原理
检索记忆
查询
嵌入模型
向量相似度搜索
返回Top-K结果
保存记忆
输入
嵌入模型
向量存储
6.5.3 初始化向量存储
方式一：FAISS（本地向量库）- import faissfrom langchain.docstore import InMemoryDocstorefrom langchain_community.vectorstores import FAISSfrom langchain_openai import OpenAIEmbeddingsfrom langchain.memory import VectorStoreRetrieverMemoryfrom langchain.chains import ConversationChainfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate# 初始化嵌入模型和LLMembedding = OpenAIEmbeddings()llm = ChatOpenAI(model="gpt-4o-mini")# 创建FAISS向量存储embedding_size = 1536 # OpenAIEmbeddings维度index = faiss.IndexFlatL2(embedding_size)vectorstore = FAISS(index, embedding, InMemoryDocStore(), {})# 创建retriever（k=3表示返回最相关的3条记忆）retriever = vectorstore.as_retriever(search_kwargs={"k": 3})# 创建向量记忆memory = VectorStoreRetrieverMemory( retriever=retriever, memory_key="chat_history")# 保存一些对话记忆memory.save_context( {"input": "我最喜欢的食物是披萨"}, {"output": "好的，我记住了 你喜欢披萨"})memory.save_context( {"input": "我喜欢在周末去跑步"}, {"output": "运动是个很好的习惯！"})memory.save_context( {"input": "我最近在学习Python编程"}, {"output": "Python是一门很实用的语言，继续加油！"})# 检索相关记忆print(memory.load_memory_variables({"prompt": "我应该吃什么？"}))
- {'chat_history': 'input: 我最喜欢的食物是披萨\noutput: 好的，我记住了...'}
- from langchain_community.vectorstores import Chromafrom langchain.embeddings import OpenAIEmbeddings# 创建Chroma向量存储vectorstore = Chroma.from_documents( documents=[], # 初始为空 embedding=OpenAIEmbeddings())# 创建memorymemory = VectorStoreRetrieverMemory( retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), memory_key="history")
- from langchain_pinecone import PineconeVectorStorefrom langchain.embeddings import OpenAIEmbeddingsfrom langchain.memory import VectorStoreRetrieverMemory# 连接Pineconevectorstore = PineconeVectorStore.from_params( index_name="my-index", embedding=OpenAIEmbeddings(), namespace="memory")memory = VectorStoreRetrieverMemory( retriever=vectorstore.as_retriever(search_kwargs={"k": 5}), memory_key="long_term_memory")
将向量记忆与对话链结合：- # 创建对话链prompt = ChatPromptTemplate.from_template("""基于以下记忆回答用户问题。如果没有相关信息，请如实说明。记忆：{chat_history}问题：{input}""")conversation = ConversationChain( llm=llm, memory=memory, prompt=prompt, verbose=True)# 第一次对话response1 = conversation.invoke({"input": "我的爱好是什么？"})print(response1)# 第二次对话（跨会话）response2 = conversation.invoke({"input": "我应该学习什么编程语言？"})print(response2)
相似度搜索（默认）- retriever = vectorstore.as_retriever( search_type="similarity", # 默认 search_kwargs={"k": 3})
- retriever = vectorstore.as_retriever( search_type="mmr", # 避免返回相似度过高的结果 search_kwargs={"k": 5, "fetch_k": 20})
- retriever = vectorstore.as_retriever( search_kwargs={"k": 5, "score_threshold": 0.7})
| 适用场景 | 不适用场景 | | 跨会话持久化记忆 | 需要严格时间顺序的场景 | | 基于语义而非关键词的检索 | 短对话、简单上下文 | | 海量历史信息存储 | 实时性要求极高的场景 | | 个性化用户画像构建 | 隐私敏感的数据 |
注意事项：
k值选择：k越大，检索越多，但可能引入噪声；k越小，可能遗漏重要信息
嵌入模型：选择与场景匹配的嵌入模型，中文推荐text-embedding-3-small或中文专用模型
存储成本：云端向量数据库有存储费用，本地FAISS/Chroma免费但占用磁盘
定期清理：长期使用后需要清理无效记忆，避免检索质量下降
6.6 LangGraph中的持久化
LangGraph提供检查点机制实现状态持久化：- from langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import StateGraph, START# 创建带持久化的图checkpointer = MemorySaver()graph = StateGraph(GraphState)graph.add_node("process", process_node)graph.add_edge(START, "process")graph.compile(checkpointer=checkpointer)# 线程ID用于恢复状态config = {"configurable": {"thread_id": "user_123"}}result = graph.invoke({"input": "hello"}, config)# 恢复状态继续执行result = graph.invoke({"input": "continue"}, config)
LangGraph 的执行流程清晰简洁，核心围绕状态（State）、节点（Node）、**边（Edge）**三个概念展开：
LangGraph执行流程
State 状态
Node 节点
Edge 边
循环执行直到完成
核心概念：
| 概念 | 说明 | 关键函数 | | State | 图的共享状态，可以是dict或Pydantic模型 | StateGraph(GraphState) | | Node | 执行逻辑的函数，接收state并返回更新 | graph.add_node(name, func) | | Edge | 定义节点间的流转关系 | graph.add_edge() / graph.add_conditional_edges() | | compile | 将图编译为可执行的app | graph.compile() |
最小示例：- from langgraph.graph import StateGraph, START, END# 1. 定义状态类型class GraphState(TypedDict): input: str result: str# 2. 定义节点函数def node_a(state: GraphState) -> GraphState: return {"result": f"处理: {state['input']}"}def node_b(state: GraphState) -> GraphState: return {"result": state["result"] + " + 节点B"}# 3. 构建图graph = StateGraph(GraphState)graph.add_node("A", node_a)graph.add_node("B", node_b)graph.add_edge(START, "A") # 起点 → Agraph.add_edge("A", "B") # A → Bgraph.add_edge("B", END) # B → 终点# 4. 编译并执行app = graph.compile()result = app.invoke({"input": "hello"})print(result) # {'input': 'hello', 'result': '处理: hello + 节点B'}
用户调用 app.invoke(input) 传入初始状态
每个节点接收当前state，处理后返回更新后的state
与Chain相比，LangGraph适合需要多步骤状态传递、条件分支、人机协作的复杂场景，是构建AI Agent的核心框架。
七、总结与进阶路径
7.1 核心概念回顾
| 概念 | 作用 | 关键类 | | Model I/O | 与LLM交互的标准化接口 | ChatOpenAI, PromptTemplate, OutputParser | | Chain | 串联组件的工作流 | LLMChain, SequentialChain | | LCEL | 声明式组件组合语法 | 管道运算符 ` | | Memory | 对话状态管理 | BufferMemory, VectorStoreMemory |
7.2 2026年新特性
LCEL全面取代旧语法：v1.x推荐全部使用LCEL
LangGraph整合：复杂工作流优先使用LangGraph
7.3 进阶学习路径
LangChain基础
LCEL深入 + RAG实战
LangGraph工作流
Agent开发 + 多Agent系统
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
为什么要学习大模型？
我国在A大模型领域面临人才短缺,数量与质量均落后于发达国家。2023年，人才缺口已超百万，凸显培养不足。随着AI技术飞速发展，预计到2025年,这一缺口将急剧扩大至400万,严重制约我国AI产业的创新步伐。加强人才培养,优化教育体系,国际合作并进是破解困局、推动AI发展的关键。
大模型入门到实战全套学习大礼包
1、大模型系统化学习路线
作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！
2、大模型学习书籍&文档
学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。
3、AI大模型最新行业报告
2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。
4、大模型项目实战&配套源码
学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。
5、大模型大厂面试真题
面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。
适用人群
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
大模型 AI 能干什么？大模型是怎样获得「智能」的？用好 AI 的核心心法大模型应用业务架构大模型应用技术架构代码示例：向 GPT-3.5 灌入新知识提示工程的意义和核心思想Prompt 典型构成指令调优方法论思维链和思维树Prompt 攻击和防范…
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
为什么要做 RAG搭建一个简单的 ChatPDF检索的基础概念什么是向量表示（Embeddings）向量数据库与向量检索基于向量检索的 RAG搭建 RAG 系统的扩展知识混合检索与 RAG-Fusion 简介向量模型本地部署…
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
为什么要做 RAG什么是模型什么是模型训练求解器 & 损失函数简介小实验2：手写一个简单的神经网络并训练它什么是训练/预训练/微调/轻量化微调Transformer结构简介轻量化微调实验数据集的构建…
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
硬件选型带你了解全球大模型使用国产大模型服务搭建 OpenAI 代理热身：基于阿里云 PAI 部署 Stable Diffusion在本地计算机运行大模型大模型的私有化部署基于 vLLM 部署大模型案例：如何优雅地在阿里云私有部署开源大模型部署一套开源 LLM 项目内容安全互联网信息服务算法备案…
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
原文地址：https://blog.csdn.net/l01011_/article/details/159462254 |
广告