1. 引言

随着大语言模型(LLM)能力的快速提升,如何高效地基于大模型构建真实业务应用,成为开发者关注的核心问题。LangChain 正是在这一背景下诞生的开源框架,它通过标准化接口和可组合的组件,帮助开发者把大模型与外部数据、工具和业务流程连接起来,从而快速搭建检索问答、智能体、对话机器人等应用。

本文将从 LangChain 的核心概念讲起,逐步深入到各核心模块,并给出大量可运行的代码实战示例,帮助读者从零开始掌握 LangChain 的基本用法。

2. LangChain 是什么

LangChain 是一个用于构建大语言模型应用的开源框架,最初由 Harrison Chase 于 2022 年 10 月发起。它提供了一套统一的接口和丰富的组件库,覆盖模型调用、提示词管理、数据检索、记忆、智能体编排等应用开发全流程。

LangChain 的核心设计理念是「组合优于继承」:开发者可以把不同功能的模块像积木一样组合起来,快速构建出复杂的应用流程。同时,LangChain 对主流大模型提供商(如 OpenAI、Anthropic、Google、阿里云通义千问等)做了统一封装,使得上层业务代码可以做到模型无关,方便切换和迁移。

LangChain 生态目前主要包含以下几个部分:

  • LangChain 核心库:提供模型调用、提示词、记忆、输出解析等基础能力。
  • LangChain Community:社区贡献的第三方集成,包括各类模型、向量库、工具等。
  • LangChain Expression Language(LCEL):一种声明式组合语法,用于把多个组件串联成可执行的链。
  • LangGraph:面向有状态、多智能体复杂工作流的编排框架。
  • LangSmith:用于调试、监控和评估 LLM 应用的开发平台。

3. 环境准备与安装

在开始代码实战之前,需要先完成环境准备。本文以 Python 3.9 及以上版本为例进行演示。

3.1 安装 LangChain

使用 pip 安装 LangChain 核心库和常用集成包:

pip install langchain langchain-community langchain-openai

如果后续需要使用向量数据库和文档加载器,可以一并安装:

pip install chromadb pypdf tiktoken

3.2 配置模型访问

LangChain 本身不包含大模型,需要通过 API 或本地模型提供服务。以 OpenAI 为例,需要设置环境变量:

export OPENAI_API_KEY="sk-xxxxxx"

在 Python 代码中也可以通过参数直接传入 API Key:

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
api_key="sk-xxxxxx",
temperature=0.7
)

如果使用国内模型,例如阿里云通义千问,可以这样配置:

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="qwen-plus",
api_key="your-dashscope-api-key",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

4. 核心模块详解与实战

下面逐一介绍 LangChain 的核心模块,并给出对应的代码示例。

4.1 模型调用(Models)

LangChain 把模型分为两类:文本补全模型(LLM)和对话模型(Chat Model)。对话模型是当前的主流,它接收消息列表并返回消息对象。

from langchain_openai import ChatOpenAI
创建对话模型实例
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
直接调用
response = llm.invoke("用一句话介绍 LangChain")
print(response.content)

LangChain 还支持流式输出,适合需要逐字返回的场景:

for chunk in llm.stream("请写一首关于春天的五言绝句"):
    print(chunk.content, end="", flush=True)

4.2 提示词模板(Prompts)

提示词模板用于把用户输入和固定指令组合成完整的提示词,避免重复拼接字符串。

from langchain_core.prompts import ChatPromptTemplate
定义提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一位资深{domain}专家,请用通俗易懂的语言回答问题。"),
("human", "{question}")
])
填充模板
messages = prompt.invoke({
"domain": "大模型",
"question": "什么是 RAG?"
})
print(messages)

也可以使用更简洁的 PromptTemplate:

from langchain_core.prompts import PromptTemplate
template = PromptTemplate.from_template(
"请把下面这句话翻译成{target_language}:{sentence}"
)
prompt_text = template.format(target_language="英语", sentence="今天天气很好")
print(prompt_text)

4.3 输出解析器(Output Parsers)

输出解析器用于把模型的文本输出转换为结构化数据,方便后续程序处理。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
prompt = ChatPromptTemplate.from_template("请用一句话解释:{concept}")
使用 LCEL 把提示词、模型、解析器串联成链
chain = prompt | llm | parser
result = chain.invoke({"concept": "LangChain"})
print(result)

如果需要解析 JSON 结构,可以使用 JsonOutputParser:

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
parser = JsonOutputParser()
prompt = PromptTemplate.from_template(
"请提取下面文本中的人物姓名和年龄,以 JSON 格式返回。\n文本:{text}\n{format_instructions}",
partial_variables={"format_instructions": parser.get_format_instructions()}
)
chain = prompt | llm | parser
result = chain.invoke({"text": "张三今年28岁,李四今年32岁。"})
print(result)

4.4 记忆(Memory)

大模型本身是无状态的,每次调用都是独立的。记忆模块用于在多轮对话中保存和读取历史信息。

from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationBufferMemory(return_messages=True)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的助手。"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
保存一轮对话
memory.chat_memory.add_user_message("我叫小明")
memory.chat_memory.add_ai_message("你好小明,很高兴认识你!")
读取历史消息
history = memory.load_memory_variables({})
print(history)

更常用的做法是使用带记忆的链,LangChain 提供了 ConversationChain 简化多轮对话:

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=memory)
print(conversation.predict(input="你好,我叫小明"))
print(conversation.predict(input="我叫什么名字?"))

4.5 文档加载与切分(Document Loaders & Splitters)

处理本地文档是 RAG 应用的第一步。LangChain 提供了丰富的文档加载器,支持 PDF、Word、网页、数据库等多种来源。

from langchain_community.document_loaders import TextLoader
加载文本文件
loader = TextLoader("example.txt", encoding="utf-8")
documents = loader.load()
print(documents)

加载 PDF 文件:

from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("report.pdf")
documents = loader.load()
print(f"共加载 {len(documents)} 页")

文档加载后通常需要切分成小块,以便向量化和检索。LangChain 提供了多种切分器:

from langchain_text_splitters import RecursiveCharacterTextSplitter
text = "这是一段很长的文档内容……" * 100
splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=50,
separators=["\n\n", "\n", "。", "!", "?", " ", ""]
)
chunks = splitter.split_text(text)
print(f"切分成 {len(chunks)} 个块")

4.6 向量存储与检索(Vector Stores & Retrievers)

向量存储用于把文本块转换为向量并建立索引,是实现语义检索的基础。这里以 Chroma 为例:

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
准备文档
documents = [
"LangChain 是一个用于构建大语言模型应用的开源框架。",
"RAG 是检索增强生成的缩写,用于减少模型幻觉。",
"向量数据库用于存储和检索文本的向量表示。"
]
切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=50, chunk_overlap=10)
chunks = splitter.create_documents(documents)
创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
语义检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
results = retriever.invoke("什么是 RAG?")
for doc in results:
print(doc.page_content)

4.7 检索增强生成(RAG)实战

把上述模块组合起来,就可以构建一个完整的 RAG 问答应用。RAG 的核心流程是:先根据用户问题检索相关文档片段,再把片段和问题一起交给大模型生成答案。

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
1. 加载文档
loader = TextLoader("knowledge.txt", encoding="utf-8")
documents = loader.load()
2. 切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
chunks = splitter.split_documents(documents)
3. 构建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
4. 定义提示词模板
prompt = ChatPromptTemplate.from_template(
"""请根据以下上下文回答问题。如果上下文中没有相关信息,请直接说明不知道,不要编造。
上下文:
{context}
问题:{question}
回答:"""
)
5. 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o-mini")
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
6. 提问
answer = rag_chain.invoke("LangChain 的核心设计理念是什么?")
print(answer)

4.8 智能体(Agents)

智能体是 LangChain 中更高级的能力,它可以让大模型自主决定调用哪些工具、按什么顺序执行,从而实现更复杂的任务自动化。

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
定义一个自定义工具
@tool
def add(a: int, b: int) -> int:
"""计算两个整数的和。"""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""计算两个整数的乘积。"""
return a * b
创建模型和提示词
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个擅长数学计算的助手,请使用工具完成计算。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
创建智能体
tools = [add, multiply]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
执行任务
result = agent_executor.invoke({"input": "请计算 (23 + 45) * 2 的结果"})
print(result["output"])

5. 完整实战案例:基于 LangChain 的文档问答机器人

下面把前面介绍的知识整合起来,构建一个完整的文档问答机器人。该案例支持上传本地文档,并针对文档内容进行多轮问答。

import os
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
1. 加载并切分文档
loader = TextLoader("company_manual.txt", encoding="utf-8")
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=80)
chunks = splitter.split_documents(documents)
2. 构建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
3. 构建历史感知检索器
llm = ChatOpenAI(model="gpt-4o-mini")
contextualize_prompt = ChatPromptTemplate.from_messages([
("system", "请把用户的问题改写为不依赖历史对话的独立问题。"),
MessagesPlaceholder("chat_history"),
("human", "{input}")
])
history_aware_retriever = create_history_aware_retriever(
llm, retriever, contextualize_prompt
)
4. 构建问答链
qa_prompt = ChatPromptTemplate.from_messages([
("system", "请基于以下上下文回答问题,如果找不到答案就如实说明。\n\n{context}"),
MessagesPlaceholder("chat_history"),
("human", "{input}")
])
document_chain = create_stuff_documents_chain(llm, qa_prompt)
rag_chain = create_retrieval_chain(history_aware_retriever, document_chain)
5. 多轮问答
chat_history = []
question1 = "公司的年假政策是什么?"
response1 = rag_chain.invoke({"input": question1, "chat_history": chat_history})
print(f"Q: {question1}")
print(f"A: {response1['answer']}\n")
chat_history.extend([
HumanMessage(content=question1),
AIMessage(content=response1["answer"])
])
question2 = "那病假呢?"
response2 = rag_chain.invoke({"input": question2, "chat_history": chat_history})
print(f"Q: {question2}")
print(f"A: {response2['answer']}")

6. 常见问题与最佳实践

6.1 如何选择合适的切分策略

切分粒度直接影响检索效果。块太小会导致语义不完整,块太大会引入噪声。建议根据文档类型和问题粒度调整 chunk_size 和 chunk_overlap,一般 chunk_size 在 200 到 500 之间,chunk_overlap 设置为 chunk_size 的 10% 到 20%。

6.2 如何减少模型幻觉

在 RAG 应用中,可以通过以下方式降低幻觉:

  • 在提示词中明确要求模型只基于上下文回答,不知道就直说。
  • 提高检索质量,使用更精准的 embedding 模型和合适的 top-k。
  • 对检索结果做相关性过滤,丢弃低相关片段。

6.3 如何调试 LangChain 应用

LangChain 提供了回调机制,可以打印链的执行过程,方便排查问题:

from langchain_core.callbacks import StdOutCallbackHandler
from langchain_openai import ChatOpenAI
handler = StdOutCallbackHandler()
llm = ChatOpenAI(model="gpt-4o-mini", callbacks=[handler])
response = llm.invoke("你好")
print(response.content)

对于更复杂的应用,推荐使用 LangSmith 进行全链路追踪和评估。

7. 总结与展望

本文系统介绍了 LangChain 的核心概念和主要模块,并通过大量代码示例演示了模型调用、提示词管理、输出解析、记忆、文档处理、向量检索、RAG 和智能体等关键能力的用法。最后通过一个完整的文档问答机器人案例,展示了如何把这些模块组合成真实可用的应用。

LangChain 生态仍在快速发展中,LCEL 和 LangGraph 正在成为构建复杂工作流的主流方式。建议读者在掌握本文基础后,进一步学习 LCEL 的流式与并行能力,以及 LangGraph 对多智能体场景的支持,并结合 LangSmith 做好应用的观测与评估。

后续可以继续深入的方向包括:多模态应用、Agent 工具编排、长文本记忆管理、以及面向生产环境的性能优化与成本控制。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐