前言

在人工智能技术飞速发展的今天,大型语言模型(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 的价值主张可以概括为以下几点:

特性

说明

模块化设计

各组件高度解耦,可灵活组合

丰富的集成

支持数十种 LLM、向量数据库、工具和数据源

标准化接口

统一的抽象层,切换底层模型无需大幅修改代码

开箱即用

提供大量预构建的链和代理模板

活跃的社区

持续更新迭代,文档完善

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 的工作流程可以简述为:

  1. 索引阶段(离线):加载文档 → 分割文本 → 生成向量嵌入 → 存入向量数据库

  2. 检索阶段(在线):用户提问 → 向量化查询 → 在数据库中检索相关文档

  3. 生成阶段(在线):将检索结果与问题一起提交给 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:主要有两种策略:

  1. RAG 方案(推荐):如本文所介绍,通过向量检索只提取相关片段

  2. 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 提供的这套完整工具链,大大降低了构建智能语言应用的技术门槛。

学习路线建议:

  1. 入门阶段:掌握 LLM/ChatModel 调用、PromptTemplate、基础链构建

  2. 进阶阶段:学习 RAG 系统搭建、Memory 管理、各类工具集成

  3. 高级阶段:深入 Agent 开发、LangGraph 多代理系统、LangServe 部署

未来趋势:

随着 LangChain 的持续迭代,LangGraph 正在成为构建复杂多代理系统的主流选择。它允许开发者以图的方式定义 Agent 的状态流转,更好地处理并行执行、条件分支和循环等复杂场景。对于有志于深入 AI 工程的开发者,LangGraph 是下一个值得重点学习的方向。

人工智能应用开发正在从"能否实现"转向"如何更好地实现"。LangChain 作为这一领域的基石框架,正在帮助无数开发者将 AI 的潜力转化为真实的商业价值。希望本文能成为你探索这个激动人心领域的良好起点。


参考资源


如果你在学习或实践 LangChain 的过程中遇到任何问题,欢迎在评论区交流探讨。技术的进步源于每一位开发者的思考与分享。