LangChain 入门指导:从零开始构建智能语言应用
前言
在人工智能技术飞速发展的今天,大型语言模型(LLM)已经从实验室走向了真实的生产环境。然而,对于大多数开发者而言,如何将 GPT-4、Claude、Llama 等强大的语言模型与实际业务场景有机结合,仍然是一项充满挑战的工作。单纯调用模型 API 虽然简单,但要构建一个功能完善、逻辑复杂的智能应用,往往需要大量的工程化工作——链式调用、记忆管理、工具集成、向量检索……每一个环节都需要开发者从头设计和实现。
正是在这样的背景下,LangChain 应运而生。作为目前最受欢迎的 LLM 应用开发框架之一,LangChain 提供了一套完整的工具集和抽象层,帮助开发者快速构建基于语言模型的智能应用。无论你是想搭建一个智能客服系统、文档问答助手,还是自动化代码审查工具,LangChain 都能为你提供强有力的支撑。
本文将系统性地介绍 LangChain 的核心概念、主要模块和实际使用方法,帮助你从零开始掌握这一强大的开发框架,并为日后构建复杂的 AI 应用打下坚实基础。
一、什么是 LangChain?
1.1 框架简介
LangChain 是由 Harrison Chase 于 2022 年 10 月发布的开源框架,最初以 Python 库的形式出现,随后迅速扩展到 JavaScript/TypeScript 生态。它的核心目标是:让开发者能够轻松地将语言模型与其他数据源和计算资源连接起来,构建具有实际应用价值的智能系统。
截至 2024 年,LangChain 已经成为 GitHub 上增长最快的开源项目之一,拥有数万颗星标,社区活跃度极高。众多企业和开发者正在使用 LangChain 构建各类 AI 应用,包括但不限于:
智能问答系统:基于企业私有文档的知识库问答
对话式 AI 助手:具备记忆能力的多轮对话机器人
自动化代理(Agent):能够自主规划和执行任务的 AI 代理
数据分析助手:通过自然语言与数据库或数据文件交互
代码生成与审查工具:辅助开发者编写、优化代码
1.2 为什么选择 LangChain?
在没有 LangChain 之前,开发者如果想构建一个能够"记住"对话历史、同时检索外部文档、并在必要时调用计算工具的智能助手,需要自行编写大量的"胶水代码"。这不仅费时费力,还容易出错,且难以维护。
LangChain 的价值主张可以概括为以下几点:
1.3 LangChain 的生态系统
随着 LangChain 的发展,其生态系统也在不断扩大,目前主要包括以下几个子项目:
LangChain Core:核心抽象和基础接口
LangChain Community:第三方集成的集合
LangChain:认知架构,如链和代理
LangGraph:用于构建有状态多代理应用的图框架
LangServe:将 LangChain 应用部署为 REST API
LangSmith:用于调试、测试和监控 LLM 应用的平台
二、安装与环境配置
2.1 环境准备
在开始之前,请确保你的开发环境满足以下要求:
Python 3.8 或更高版本(推荐 3.10+)
pip 包管理器
有效的 LLM API 密钥(如 OpenAI API Key)
2.2 安装 LangChain
通过 pip 安装 LangChain 核心包:
pip install langchain
如果你打算使用 OpenAI 的模型(这是最常见的起点),还需要安装对应的集成包:
pip install langchain-openai
如果需要使用向量存储、文档加载等功能,可以安装完整的社区包:
pip install langchain-community
对于特定的功能模块,你可能还需要安装额外的依赖,例如:
# 用于向量数据库
pip install faiss-cpu chromadb
# 用于文档处理
pip install pypdf docx2txt
# 用于 Web 搜索工具
pip install duckduckgo-search
2.3 配置 API 密钥
LangChain 通过环境变量管理 API 密钥,这是一种安全的最佳实践。
方式一:直接在终端设置
export OPENAI_API_KEY="your-api-key-here"
方式二:在 Python 代码中设置(仅用于开发测试)
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
方式三:使用 .env 文件(推荐)
在项目根目录创建 .env 文件:
OPENAI_API_KEY=your-api-key-here
然后在代码中加载:
from dotenv import load_dotenv
load_dotenv()
⚠️ 安全提示:请务必将
.env文件添加到.gitignore中,避免 API 密钥泄露到代码仓库。
三、核心概念详解
LangChain 的架构围绕几个核心抽象展开,理解这些概念是掌握整个框架的关键。
3.1 模型(Models)
模型是 LangChain 的基础组件,是整个框架的"大脑"。LangChain 对不同类型的模型进行了统一的抽象,主要分为两类:
3.1.1 LLM(Large Language Model)
传统的文本补全模型,接受字符串输入,返回字符串输出。
from langchain_openai import OpenAI
# 初始化 LLM
llm = OpenAI(
model="gpt-3.5-turbo-instruct",
temperature=0.7, # 控制输出的随机性,0表示确定性输出
max_tokens=500
)
# 调用模型
response = llm.invoke("请用一句话介绍人工智能的发展历史")
print(response)
3.1.2 ChatModel(对话模型)
专为多轮对话设计的模型,接受消息列表作为输入,返回消息对象。这是目前更为主流的使用方式。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
# 初始化 ChatModel
chat = ChatOpenAI(
model="gpt-4o",
temperature=0.7
)
# 构建消息列表
messages = [
SystemMessage(content="你是一位专业的 Python 编程导师,擅长用简洁易懂的方式解释复杂概念。"),
HumanMessage(content="什么是装饰器(Decorator)?能举个例子吗?")
]
# 调用模型
response = chat.invoke(messages)
print(response.content)
LangChain 支持的模型提供商非常丰富,包括:
OpenAI:GPT-4、GPT-3.5-turbo 等
Anthropic:Claude 3 系列
Google:Gemini 系列
Meta:Llama 系列(通过 Ollama 等方式本地运行)
Hugging Face:各类开源模型
百度文心、阿里通义千问、智谱 AI 等国内模型
3.2 提示词模板(Prompt Templates)
Prompt Template 是 LangChain 中用于标准化提示词构建的核心工具。它允许你定义带有变量占位符的模板,然后在运行时动态填充具体内容。
3.2.1 基础提示词模板
from langchain_core.prompts import PromptTemplate
# 定义模板
template = """
你是一位经验丰富的{role}。
请根据以下要求,为用户提供专业建议:
用户问题:{question}
请用中文回答,回答应简洁、专业、具有可操作性。
"""
# 创建 PromptTemplate 对象
prompt = PromptTemplate(
input_variables=["role", "question"],
template=template
)
# 格式化提示词
formatted_prompt = prompt.format(
role="软件架构师",
question="如何在微服务架构中处理分布式事务?"
)
print(formatted_prompt)
3.2.2 对话提示词模板
from langchain_core.prompts import ChatPromptTemplate
# 创建对话提示词模板
chat_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的{language}代码助手,请用{language}回答问题。"),
("human", "请帮我解释以下代码:\n{code}"),
])
# 格式化
messages = chat_prompt.format_messages(
language="Python",
code="""
def fibonacci(n):
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
"""
)
print(messages)
3.2.3 Few-Shot 提示词模板
Few-Shot 提示是通过提供少量示例来引导模型输出特定格式或风格的技术:
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
# 定义示例
examples = [
{
"input": "苹果",
"output": "苹果是一种常见的水果,富含维生素C和膳食纤维,口感甜脆,常见颜色有红色、绿色和黄色。"
},
{
"input": "香蕉",
"output": "香蕉是一种热带水果,富含钾元素和天然糖分,口感软糯香甜,是优质的能量补充食品。"
}
]
# 示例模板
example_prompt = PromptTemplate(
input_variables=["input", "output"],
template="输入:{input}\n输出:{output}"
)
# 创建 FewShot 模板
few_shot_prompt = FewShotPromptTemplate(
examples=examples,
example_prompt=example_prompt,
prefix="请按照以下示例的格式,描述给定的水果:",
suffix="输入:{fruit}\n输出:",
input_variables=["fruit"]
)
print(few_shot_prompt.format(fruit="草莓"))
3.3 输出解析器(Output Parsers)
语言模型的输出是非结构化的文本,在实际应用中,我们往往需要将其解析为特定的数据格式。LangChain 提供了多种输出解析器来处理这一需求。
3.3.1 字符串输出解析器
最简单的解析器,直接返回字符串:
from langchain_core.output_parsers import StrOutputParser
parser = StrOutputParser()
# 将在链中使用
3.3.2 JSON 输出解析器
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
# 定义期望的输出结构
class MovieReview(BaseModel):
title: str = Field(description="电影标题")
rating: float = Field(description="评分,满分10分")
summary: str = Field(description="简短的影评摘要")
pros: list[str] = Field(description="优点列表")
cons: list[str] = Field(description="缺点列表")
# 创建解析器
parser = JsonOutputParser(pydantic_object=MovieReview)
# 创建提示词
prompt = PromptTemplate(
template="请对电影《{movie}》进行评价。\n{format_instructions}\n",
input_variables=["movie"],
partial_variables={"format_instructions": parser.get_format_instructions()}
)
# 组合链
model = ChatOpenAI(temperature=0)
chain = prompt | model | parser
# 运行
result = chain.invoke({"movie": "星际穿越"})
print(result)
3.4 链(Chains)
链是 LangChain 最核心的概念之一,它将多个组件按照一定的逻辑顺序连接起来,形成一个处理流程。
3.4.1 LCEL(LangChain Expression Language)
LangChain 在 0.1 版本后引入了 LCEL(LangChain 表达式语言),使用管道符 | 将组件连接起来,语法简洁优雅:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 定义各组件
prompt = ChatPromptTemplate.from_template(
"请用三点总结{topic}的核心优势,每点不超过50字。"
)
model = ChatOpenAI(model="gpt-4o", temperature=0)
parser = StrOutputParser()
# 使用 | 连接组件,构建链
chain = prompt | model | parser
# 调用链
result = chain.invoke({"topic": "微服务架构"})
print(result)
3.4.2 顺序链(Sequential Chain)
将多个步骤串联起来,前一步的输出作为后一步的输入:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI(temperature=0.7)
parser = StrOutputParser()
# 第一步:生成文章大纲
outline_prompt = ChatPromptTemplate.from_template(
"请为主题「{topic}」生成一份包含5个要点的文章大纲。"
)
# 第二步:根据大纲扩展内容
expand_prompt = ChatPromptTemplate.from_template(
"根据以下大纲,为每个要点写一段详细的说明(每段约100字):\n{outline}"
)
# 构建顺序链
chain = (
outline_prompt
| model
| parser
| (lambda outline: {"outline": outline})
| expand_prompt
| model
| parser
)
result = chain.invoke({"topic": "Python 异步编程"})
print(result)
3.5 记忆(Memory)
对话系统中,"记忆"至关重要——它让 AI 助手能够记住之前的对话内容,实现真正的多轮交互。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# 创建模型和提示词
model = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的 AI 助手,请用中文回答问题。"),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
# 存储会话历史的字典
store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# 构建带记忆的链
chain = prompt | model
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
# 第一轮对话
config = {"configurable": {"session_id": "user_001"}}
response1 = chain_with_history.invoke(
{"input": "我叫张伟,是一名后端开发工程师。"},
config=config
)
print("助手:", response1.content)
# 第二轮对话(模型应该记得用户的名字和职业)
response2 = chain_with_history.invoke(
{"input": "你还记得我是做什么工作的吗?"},
config=config
)
print("助手:", response2.content)
3.6 文档加载器(Document Loaders)
在实际应用中,我们经常需要让 AI 处理各种格式的文档。LangChain 提供了丰富的文档加载器:
from langchain_community.document_loaders import (
TextLoader, # 加载 .txt 文件
PyPDFLoader, # 加载 PDF 文件
CSVLoader, # 加载 CSV 文件
WebBaseLoader, # 加载网页内容
DirectoryLoader # 加载整个目录
)
# 加载 PDF 文件
pdf_loader = PyPDFLoader("./documents/report.pdf")
pdf_docs = pdf_loader.load()
print(f"加载了 {len(pdf_docs)} 页 PDF 内容")
print(f"第一页内容预览:{pdf_docs[0].page_content[:200]}")
# 加载网页
web_loader = WebBaseLoader("https://python.langchain.com/docs/introduction/")
web_docs = web_loader.load()
print(f"网页标题:{web_docs[0].metadata.get('title', '未知')}")
# 加载整个目录的文本文件
dir_loader = DirectoryLoader("./docs/", glob="**/*.txt", loader_cls=TextLoader)
dir_docs = dir_loader.load()
print(f"目录中共加载了 {len(dir_docs)} 个文档")
3.7 文本分割器(Text Splitters)
由于语言模型存在上下文长度限制,我们需要将长文档切分成更小的块(chunks)进行处理:
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
CharacterTextSplitter
)
# 推荐使用 RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每个块的最大字符数
chunk_overlap=200, # 相邻块之间的重叠字符数(保证上下文连贯)
length_function=len,
separators=["\n\n", "\n", "。", "!", "?", " ", ""]
)
# 对文档进行分割
splits = text_splitter.split_documents(pdf_docs)
print(f"文档被分割为 {len(splits)} 个块")
print(f"第一个块的内容:\n{splits[0].page_content}")
print(f"块的元数据:{splits[0].metadata}")
3.8 向量存储(Vector Stores)与检索(Retrieval)
向量存储是构建 RAG(检索增强生成)系统的核心组件。它将文本转换为向量表示,并支持语义相似度搜索。
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
# 1. 加载文档
loader = TextLoader("./knowledge_base.txt", encoding="utf-8")
documents = loader.load()
# 2. 分割文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
splits = text_splitter.split_documents(documents)
# 3. 创建嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 4. 创建向量数据库并存储文档
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db" # 持久化存储路径
)
# 5. 创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity", # 相似度搜索
search_kwargs={"k": 5} # 返回最相似的 5 个结果
)
# 6. 执行检索
query = "如何配置系统的安全策略?"
relevant_docs = retriever.invoke(query)
for i, doc in enumerate(relevant_docs):
print(f"\n--- 相关文档 {i+1} ---")
print(doc.page_content)
四、实战案例:构建 RAG 问答系统
掌握了上述基础概念后,让我们来构建一个完整的 RAG(Retrieval-Augmented Generation,检索增强生成) 系统。这是 LangChain 最经典的应用场景之一,广泛用于企业知识库问答、文档智能分析等场景。
4.1 RAG 的基本原理
RAG 的工作流程可以简述为:
索引阶段(离线):加载文档 → 分割文本 → 生成向量嵌入 → 存入向量数据库
检索阶段(在线):用户提问 → 向量化查询 → 在数据库中检索相关文档
生成阶段(在线):将检索结果与问题一起提交给 LLM → 生成最终回答
4.2 完整代码实现
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
class RAGSystem:
"""基于 LangChain 的 RAG 问答系统"""
def __init__(self, model_name: str = "gpt-4o"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.vectorstore = None
self.retriever = None
def load_and_index(self, file_paths: list[str]):
"""加载文档并建立索引"""
all_documents = []
for file_path in file_paths:
print(f"正在加载文档:{file_path}")
if file_path.endswith(".pdf"):
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
all_documents.extend(docs)
print(f"共加载 {len(all_documents)} 个文档片段")
# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=150,
separators=["\n\n", "\n", "。", ";", " ", ""]
)
splits = text_splitter.split_documents(all_documents)
print(f"文档分割完成,共 {len(splits)} 个文本块")
# 建立向量索引
print("正在构建向量索引,请稍候...")
self.vectorstore = Chroma.from_documents(
documents=splits,
embedding=self.embeddings,
persist_directory="./rag_chroma_db"
)
# 创建检索器
self.retriever = self.vectorstore.as_retriever(
search_type="mmr", # 使用 MMR 算法,提高结果多样性
search_kwargs={"k": 6, "fetch_k": 20}
)
print("索引构建完成!系统已就绪。")
def build_chain(self):
"""构建 RAG 处理链"""
# 设计 RAG 提示词模板
rag_prompt = ChatPromptTemplate.from_template("""
你是一个专业的知识库问答助手。请根据提供的上下文信息,准确、完整地回答用户的问题。
**重要提示:**
- 只根据提供的上下文信息进行回答
- 如果上下文中没有相关信息,请明确告知用户
- 回答应该简洁清晰,必要时可以列出要点
- 如果需要引用具体内容,请指明来源
---
上下文信息:
{context}
---
用户问题:{question}
回答:
""")
def format_docs(docs):
"""格式化检索到的文档"""
formatted = []
for i, doc in enumerate(docs):
source = doc.metadata.get("source", "未知来源")
page = doc.metadata.get("page", "")
source_info = f"[来源: {source}" + (f", 第{page+1}页" if page != "" else "") + "]"
formatted.append(f"{source_info}\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
# 构建 RAG 链
rag_chain = (
{
"context": self.retriever | format_docs,
"question": RunnablePassthrough()
}
| rag_prompt
| self.llm
| StrOutputParser()
)
return rag_chain
def query(self, question: str) -> str:
"""执行问答查询"""
if not self.retriever:
raise ValueError("请先调用 load_and_index() 方法加载并索引文档")
chain = self.build_chain()
return chain.invoke(question)
def query_with_sources(self, question: str) -> dict:
"""执行问答查询并返回来源信息"""
if not self.retriever:
raise ValueError("请先调用 load_and_index() 方法加载并索引文档")
# 检索相关文档
relevant_docs = self.retriever.invoke(question)
# 构建链并获取答案
chain = self.build_chain()
answer = chain.invoke(question)
# 整理来源信息
sources = []
for doc in relevant_docs:
source = {
"file": doc.metadata.get("source", "未知"),
"page": doc.metadata.get("page", "N/A"),
"preview": doc.page_content[:150] + "..."
}
sources.append(source)
return {
"question": question,
"answer": answer,
"sources": sources
}
# 使用示例
if __name__ == "__main__":
# 初始化 RAG 系统
rag = RAGSystem(model_name="gpt-4o")
# 加载文档(支持多个文件)
rag.load_and_index([
"./docs/company_policy.pdf",
"./docs/product_manual.txt",
"./docs/faq.pdf"
])
# 简单问答
answer = rag.query("公司的请假政策是什么?")
print(f"回答:{answer}")
# 带来源信息的问答
result = rag.query_with_sources("产品如何进行安全配置?")
print(f"\n问题:{result['question']}")
print(f"回答:{result['answer']}")
print(f"\n参考来源:")
for i, source in enumerate(result['sources'][:3]):
print(f" {i+1}. {source['file']} - 第{source['page']}页")
print(f" 摘要:{source['preview']}")
五、进阶主题:构建 AI 代理(Agent)
如果说链(Chain)是按照预定路径执行的"流程图",那么代理(Agent)则是能够自主决策、动态选择工具、迭代执行任务的"智能体"。
5.1 Agent 的工作原理
Agent 的核心思路是 ReAct(Reasoning + Acting) 框架:
[用户输入]
↓
[思考:我需要做什么?]
↓
[决策:使用哪个工具?]
↓
[行动:调用工具]
↓
[观察:工具返回了什么?]
↓
[判断:任务完成了吗?]
├─ 未完成 → 返回"思考"步骤
└─ 完成 → 输出最终答案
5.2 创建一个简单的 Agent
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.tools import tool
from langchain import hub
import datetime
import math
# 定义工具函数
@tool
def calculate(expression: str) -> str:
"""
执行数学计算。输入一个数学表达式字符串,返回计算结果。
支持基本运算、三角函数、对数等。
示例:'2 + 3 * 4', 'math.sqrt(16)', 'math.sin(math.pi/2)'
"""
try:
# 提供安全的数学命名空间
safe_dict = {
"math": math,
"__builtins__": {}
}
result = eval(expression, safe_dict)
return f"计算结果:{result}"
except Exception as e:
return f"计算错误:{str(e)}"
@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""
获取当前时间和日期。
参数 timezone 指定时区,默认为上海时区(中国标准时间)。
"""
now = datetime.datetime.now()
return f"当前时间({timezone}):{now.strftime('%Y年%m月%d日 %H:%M:%S')}"
@tool
def word_counter(text: str) -> str:
"""
统计文本的字数、字符数和段落数。
输入一段文本,返回详细的统计信息。
"""
char_count = len(text)
word_count = len(text.split())
paragraph_count = len([p for p in text.split('\n\n') if p.strip()])
return f"""文本统计结果:
- 字符数(含空格):{char_count}
- 单词/词语数:{word_count}
- 段落数:{paragraph_count}"""
# 工具列表
tools = [calculate, get_current_time, word_counter]
# 初始化模型(Agent 建议使用能力更强的模型)
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 从 Hub 加载 ReAct 提示词模板
prompt = hub.pull("hwchase17/react")
# 创建 Agent
agent = create_react_agent(llm, tools, prompt)
# 创建 Agent 执行器
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 显示详细的推理过程
max_iterations=10, # 最大迭代次数,防止无限循环
handle_parsing_errors=True
)
# 运行 Agent
result = agent_executor.invoke({
"input": "现在几点了?另外,请帮我计算 (15 * 8 + 32) / 4 的结果,并告诉我 sin(90°) 的值(用弧度计算)。"
})
print("\n最终答案:", result["output"])
5.3 使用预构建工具
LangChain 提供了大量开箱即用的工具,可以直接集成到 Agent 中:
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.tools.wikipedia.tool import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
# 网络搜索工具
search_tool = DuckDuckGoSearchRun()
# Wikipedia 查询工具
wikipedia = WikipediaQueryRun(
api_wrapper=WikipediaAPIWrapper(
lang="zh", # 使用中文维基百科
top_k_results=3
)
)
# 将这些工具加入到 Agent 中
tools = [search_tool, wikipedia, calculate, get_current_time]
六、最佳实践与注意事项
6.1 成本控制
大量调用 LLM API 可能产生较高费用,以下是一些控制成本的建议:
使用缓存:对于重复查询,启用 LangChain 内置的缓存机制
选择合适的模型:并非所有任务都需要最强的模型,简单任务可使用 GPT-3.5
优化 Prompt:精简提示词,减少不必要的 token 消耗
设置 max_tokens 限制:避免模型生成过长的不必要内容
from langchain_community.cache import InMemoryCache
from langchain_core.globals import set_llm_cache
# 启用内存缓存
set_llm_cache(InMemoryCache())
# 对于更持久的缓存,可以使用 SQLite
from langchain_community.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))
6.2 错误处理与重试
from langchain_openai import ChatOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
# LangChain 原生支持重试配置
llm = ChatOpenAI(
model="gpt-4o",
max_retries=3, # 最大重试次数
request_timeout=60 # 请求超时时间(秒)
)
# 在链中添加错误处理
from langchain_core.runnables import RunnableLambda
def handle_error(error):
return f"处理请求时发生错误:{str(error)}"
safe_chain = chain.with_fallbacks(
fallbacks=[RunnableLambda(lambda x: "服务暂时不可用,请稍后重试")],
exception_key="error"
)
6.3 流式输出
对于需要实时显示生成内容的场景(如聊天界面),可以使用流式输出:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o", streaming=True)
prompt = ChatPromptTemplate.from_template("请写一篇关于{topic}的短文。")
chain = prompt | llm | StrOutputParser()
# 流式输出
print("正在生成内容:")
for chunk in chain.stream({"topic": "人工智能的未来发展"}):
print(chunk, end="", flush=True)
print("\n\n生成完成!")
6.4 调试与追踪
使用 LangSmith 进行追踪(推荐在生产环境中使用):
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-project"
# 设置后,所有 LangChain 调用都会自动追踪
简单的本地调试:
from langchain.globals import set_debug, set_verbose
# 开启详细模式(显示输入输出)
set_verbose(True)
# 开启调试模式(显示所有内部信息)
set_debug(True)
七、常见问题解答(FAQ)
Q1:LangChain 支持哪些中文模型?
A:LangChain 通过 langchain-community 包支持多款国内主流模型,包括:
百度文心一言(ERNIE-Bot)
阿里通义千问(Qwen)
智谱 AI(ChatGLM)
讯飞星火(Spark)
MiniMax 等
Q2:如何处理超长文档,使其不超过模型的上下文限制?
A:主要有两种策略:
RAG 方案(推荐):如本文所介绍,通过向量检索只提取相关片段
Map-Reduce 方案:将文档分块分别处理,最后汇总结果
Q3:LangChain 与 LlamaIndex 有什么区别?
A:两者都是 LLM 应用开发框架,但侧重点不同:
LangChain:更注重 Agent 编排、工具调用、链式处理,应用场景更广泛
LlamaIndex(原 GPT Index):更专注于数据索引和检索,在 RAG 场景下功能更专深
在实际项目中,两者可以结合使用。
Q4:如何在本地运行开源模型(不依赖 OpenAI)?
A:推荐使用 Ollama:
# 安装 Ollama(访问 ollama.ai 下载)
# 拉取模型
ollama pull llama3.2
# 在 LangChain 中使用
from langchain_community.llms import Ollama
llm = Ollama(model="llama3.2")
response = llm.invoke("你好,请介绍一下你自己。")
print(response)
八、总结与展望
通过本文的系统介绍,我们从 LangChain 的基本概念出发,逐步深入到模型调用、提示词设计、链式处理、记忆管理、文档检索,直至完整的 RAG 系统和 Agent 构建。LangChain 提供的这套完整工具链,大大降低了构建智能语言应用的技术门槛。
学习路线建议:
入门阶段:掌握 LLM/ChatModel 调用、PromptTemplate、基础链构建
进阶阶段:学习 RAG 系统搭建、Memory 管理、各类工具集成
高级阶段:深入 Agent 开发、LangGraph 多代理系统、LangServe 部署
未来趋势:
随着 LangChain 的持续迭代,LangGraph 正在成为构建复杂多代理系统的主流选择。它允许开发者以图的方式定义 Agent 的状态流转,更好地处理并行执行、条件分支和循环等复杂场景。对于有志于深入 AI 工程的开发者,LangGraph 是下一个值得重点学习的方向。
人工智能应用开发正在从"能否实现"转向"如何更好地实现"。LangChain 作为这一领域的基石框架,正在帮助无数开发者将 AI 的潜力转化为真实的商业价值。希望本文能成为你探索这个激动人心领域的良好起点。
参考资源
如果你在学习或实践 LangChain 的过程中遇到任何问题,欢迎在评论区交流探讨。技术的进步源于每一位开发者的思考与分享。