作者:互联网 时间: 2026-07-21 08:57:54
本文基于 2026年 LangChain 官方最新稳定版(v1.x) 重写,完全对标官网 Concepts 权威定义,摒弃所有废弃 API、修正旧版概念错误。全文严格采用 LangChain 六大核心组件 标准体系,所有代码为生产级可直接运行写法,适配 Python3.9–3.12,适合入门学习、项目开发与面试备考。

LangChain 是面向大语言模型(LLM)的生产级AI应用编排框架,核心解决 LLM 无记忆、无外部知识、无法自主执行任务的短板,可快速搭建对话机器人、知识库问答(RAG)、智能Agent、自动化工作流等应用。
LangChain 所有应用均由以下六大核心组件组合构建,为官网固定标准,无增减、无合并:
# 核心框架pip install langchain langchain-core langgraph# 生态与模型适配pip install langchain-community langchain-openai# 工具依赖pip install faiss-cpu python-dotenv pydantic
项目根目录新建 .env 文件,统一管理密钥与配置:
OPENAI_API_KEY=你的密钥OPENAI_BASE_URL=模型袋里地址(可选)LANGCHAIN_TRACING_V2=trueLANGCHAIN_API_KEY=你的LangSmith密钥LANGCHAIN_PROJECT=LangChain-v1.x-Demo
本章为教程核心,逐一对标官方定义,包含组件原理、核心作用、适用场景、生产级代码。
Models 是 LangChain 最底层核心组件,统一全球所有大模型的调用协议,屏蔽不同厂商 API 差异,支持一键切换模型。v1.x 分为三类标准模型:对话模型、补全模型、嵌入模型。
所有模型通用:invoke()单次调用、stream()流式输出、batch()批量调用,同时支持异步方法。
from dotenv import load_dotenvimport osfrom langchain_openai import ChatOpenAI, OpenAI, OpenAIEmbeddingsload_dotenv()# 1. 初始化对话模型(生产首选)chat_model = ChatOpenAI(model="gpt-4o-mini",temperature=0.7,max_tokens=2048,api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_BASE_URL"),streaming=True)# 2. 初始化文本补全模型llm = OpenAI(model="gpt-3.5-turbo-instruct", temperature=0.5)# 3. 初始化嵌入模型embedding = OpenAIEmbeddings()if __name__ == "__main__":# 单次调用res = chat_model.invoke("简述LangChain六大核心组件")print("单次调用结果:", res.content)# 流式输出print("n流式输出:")for chunk in chat_model.stream("Models组件的核心作用"):print(chunk.content, end="", flush=True)
Prompt 组件用于解决硬编码提示词不可复用、格式混乱、无法动态传参的问题。通过模板封装系统指令、用户问题、历史上下文,实现提示词统一管理、批量复用、动态渲染。
from langchain_core.prompts import ChatPromptTemplate# 构建多角色可复用提示模板prompt = ChatPromptTemplate.from_messages([("system", "你是LangChain技术专家,回答简洁专业,适配初学者"),("human", "请详细讲解{component}的核心作用与使用场景")])# 动态传参渲染模板format_prompt = prompt.format(component="Prompts提示模板组件")print("渲染后提示词:n", format_prompt)# LCEL组合调用chain = prompt | chat_modelres = chain.invoke({"component": "Output Parsers输出解析组件"})print("n模型回复:n", res.content)
大模型默认输出非结构化自由文本,无法对接程序逻辑、数据库、前端渲染。Output Parsers 核心作用是约束并解析模型输出格式,将文本转为 JSON、列表、自定义对象等结构化数据,是业务落地的必备组件。
from langchain_core.output_parsers import JsonOutputParserfrom pydantic import BaseModel, Fieldfrom typing import List# 自定义结构化数据模型class LangChainComponent(BaseModel):component_name: str = Field(description="组件名称")core_function: str = Field(description="核心功能")application_scene: List[str] = Field(description="适用场景")# 初始化解析器parser = JsonOutputParser(pydantic_object=LangChainComponent)# 带格式约束的提示词prompt = ChatPromptTemplate.from_messages([("system", "严格输出JSON格式,无多余内容"),("human", "介绍LangChain六大核心组件中的Memory组件:{format_instructions}")]).partial(format_instructions=parser.get_format_instructions())# 完整链路chain = prompt | chat_model | parserres = chain.invoke({})print("结构化解析结果:n", res)print("适用场景:n", res["application_scene"])
LLM 无原生上下文记忆,单次请求相互独立。Memory 组件负责存储、管理、迭代对话历史,实现连续多轮对话。v1.x 废弃全局记忆,统一使用会话级隔离记忆,支持多用户并发、上下文持久化。
from langchain_core.runnables.history import RunnableWithMessageHistoryfrom langchain_community.chat_message_histories import ChatMessageHistory# 会话存储(生产可替换Redis/数据库)session_store = {}# 获取会话历史def get_session(session_id: str) -> ChatMessageHistory:if session_id not in session_store:session_store[session_id] = ChatMessageHistory()return session_store[session_id]# 基础对话链路prompt = ChatPromptTemplate.from_messages([("system", "你是AI助手,记住用户对话信息,连贯回答"),("human", "{input}")])base_chain = prompt | chat_model# 包装会话记忆memory_chain = RunnableWithMessageHistory(runnable=base_chain,get_session_history=get_session,input_messages_key="input")# 多会话隔离测试if __name__ == "__main__":memory_chain.invoke({"input": "我叫小明"}, config={"session_id": "user_001"})print("用户001提问结果:", memory_chain.invoke({"input": "我叫什么名字?"}, config={"session_id": "user_001"}).content)print("用户002提问结果:", memory_chain.invoke({"input": "我叫什么名字?"}, config={"session_id": "user_002"}).content)
Documents 是官方独立核心组件,为 RAG 提供原始数据预处理能力。负责加载各类外部数据、清洗冗余内容、智能切片,将异构数据统一封装为标准 Document 对象,是所有知识库应用的前置基础。
page_content 内容 + metadata 元数据结构from langchain_core.documents import Documentfrom langchain_community.document_loaders import TextLoaderfrom langchain.text_splitter import RecursiveCharacterTextSplitter# 1. 手动创建标准文档对象doc = Document(page_content="LangChain六大核心组件:Models、Prompts、Output Parsers、Memory、Documents、Retrieval",metadata={"source": "官方文档", "version": "v1.x"})print("标准文档内容:", doc.page_content)print("文档元数据:", doc.metadata)# 2. 加载本地文件loader = TextLoader("test.txt", encoding="utf-8")raw_docs = loader.load()# 3. 智能文本切片splitter = RecursiveCharacterTextSplitter(chunk_size=500,chunk_overlap=50,separators=["nn", "n", "。"])split_docs = splitter.split_documents(raw_docs)print(f"n切片后文档片段数量:{len(split_docs)}")
Retrieval 是六大组件最后一环,承接 Documents 预处理后的标准化文档,完成向量化存储、语义召回、知识增强问答。同时集成工具调用与智能Agent能力,支持模型自主调用外部工具、自主决策执行任务,是实现RAG知识库与自动化智能体的核心。
from langchain_community.vectorstores import FAISSfrom langchain.chains import RetrievalQA# 承接上文文档切片与嵌入模型vector_db = FAISS.from_documents(split_docs, embedding)retriever = vector_db.as_retriever(search_type="similarity", k=3)# 构建RAG问答链路rag_chain = RetrievalQA.from_chain_type(llm=chat_model,retriever=retriever,return_source_documents=True)# 问答测试res = rag_chain.invoke("LangChain六大核心组件是什么?")print("RAG问答结果:", res["result"])
from langchain.tools import toolfrom langgraph.prebuilt import create_react_agent# 自定义工具def calculator(express: str) -> str:"""数学计算工具,用于处理所有数学运算"""return str(eval(express))# 创建官方标准Agentagent = create_react_agent(chat_model, tools=[calculator])# 自主执行任务res = agent.invoke({"messages": [("human", "计算 256*18+720 的结果")]})print("Agent执行结果:", res["messages"][-1].content)
LCEL(LangChain Expression Language)是 v1.x 唯一官方组件组合语法,通过 | 管道符串联所有Runnable组件,替代老旧Chain写法,具备极简、异步、流式、可观测、可嵌套的优势。
# 标准链路:提示词 -> 模型 -> 解析器chain = ChatPromptTemplate.from_messages([("system", "精简回答用户问题"),("human", "{query}")]) | chat_model | JsonOutputParser()# 批量调用res = chain.batch([{"query": "什么是LCEL"}, {"query": "什么是RAG"}])print(res)
整合六大核心组件,实现企业级轻量化知识库问答系统,支持多轮记忆、私有文档检索、结构化输出。
from dotenv import load_dotenvimport osfrom langchain_openai import ChatOpenAI, OpenAIEmbeddingsfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain_community.document_loaders import TextLoaderfrom langchain_community.vectorstores import FAISSfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables.history import RunnableWithMessageHistoryfrom langchain_community.chat_message_histories import ChatMessageHistoryfrom langchain_core.output_parsers import JsonOutputParserfrom langchain_core.runnables import RunnableParallel, RunnablePassthroughload_dotenv()# 1. 初始化基础组件llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=os.getenv("OPENAI_API_KEY"))embeddings = OpenAIEmbeddings()session_store = {}# 2. 会话记忆def get_session(sid):return session_store.setdefault(sid, ChatMessageHistory())# 3. 文档处理(Documents组件)loader = TextLoader("langchain_doc.txt", encoding="utf-8")docs = loader.load()split_docs = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=60).split_documents(docs)# 4. 向量库与检索(Retrieval组件)vector_db = FAISS.from_documents(split_docs, embeddings)retriever = vector_db.as_retriever(k=3)# 5. 提示模板(Prompts组件)rag_prompt = ChatPromptTemplate.from_messages([("system", "基于参考文档精准回答,结合上下文连贯回复:{context}"),("human", "{question}")])# 6. LCEL链路编排rag_chain = RunnableParallel(context=lambda x: retriever.invoke(x["question"]),question=RunnablePassthrough()) | rag_prompt | llm# 7. 包装记忆链路chat_rag = RunnableWithMessageHistory(rag_chain, get_session, input_messages_key="question")# 8. 项目调用if __name__ == "__main__":result = chat_rag.invoke({"question": "LangChain六大核心组件分别是什么?"},config={"session_id": "final_demo_001"})print("最终问答结果:n", result.content)