LangChain入门与实践

1. 引言

随着大语言模型（LLM）技术的快速发展，如何有效地构建基于LLM的应用程序成为了开发者面临的重要挑战。LangChain作为一个专门为此设计的开源框架，为开发者提供了一套完整的工具链，使得构建复杂的AI应用变得更加简单和高效。

2. LangChain概述

2.1 什么是LangChain

LangChain是一个开源框架，专门用于开发由大语言模型（LLM）驱动的应用程序。它提供了一套标准化的接口、工具和端到端的链式调用机制，使开发者能够轻松地将LLM集成到各种应用场景中。

graph TD A[开发者] --> B[LangChain框架] B --> C[大语言模型] B --> D[外部数据源] B --> E[工具集成] C --> F[智能应用] D --> F E --> F

2.2 核心设计理念

LangChain的设计基于以下几个核心理念：

mindmap root((LangChain核心理念)) 模块化设计独立使用组合使用松耦合架构标准化接口统一调用方式多LLM支持兼容性保证链式组合工作流构建组件串联复杂任务分解数据感知外部数据源实时数据处理多格式支持代理能力工具交互环境感知自主决策

2.3 主要特性

多模型支持：支持OpenAI、Anthropic、Hugging Face等多种LLM
丰富的组件库：提供提示模板、链、代理、记忆等核心组件
灵活的数据处理：支持多种文档格式和数据源
强大的检索能力：集成向量数据库和检索算法
可扩展性：易于添加自定义组件和集成

3. LangChain架构详解

3.1 整体架构

LangChain采用分层架构设计，从底层的模型接口到上层的应用逻辑，形成了一个完整的技术栈：

graph TB subgraph "应用层" A1[智能问答系统] A2[文档分析工具] A3[代码生成器] A4[智能助手] end subgraph "框架层" B1[Chains 链式调用] B2[Agents 智能代理] B3[Memory 记忆管理] B4[Tools 工具集成] end subgraph "组件层" C1[Prompts 提示模板] C2[Document Loaders 文档加载] C3[Text Splitters 文本分割] C4[Embeddings 向量化] C5[Vector Stores 向量存储] C6[Retrievers 检索器] end subgraph "模型层" D1[OpenAI GPT] D2[Anthropic Claude] D3[Hugging Face Models] D4[Local Models] end A1 --> B1 A2 --> B2 A3 --> B3 A4 --> B4 B1 --> C1 B2 --> C2 B3 --> C3 B4 --> C4 C1 --> D1 C2 --> D2 C3 --> D3 C4 --> D4

3.2 核心组件

LangChain的核心组件可以分为以下几个主要类别：

模型抽象层：统一不同LLM的调用接口
提示工程：管理和优化提示模板
链式处理：组合多个步骤形成工作流
智能代理：实现自主决策和工具调用
记忆系统：管理对话历史和上下文
数据处理：处理各种格式的文档和数据
检索系统：实现高效的信息检索

4. 核心组件深入分析

4.1 Models（模型）

Models是LangChain的基础组件，提供了与各种大语言模型交互的统一接口。

4.1.1 LLM（大语言模型）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain.llms import OpenAI, HuggingFacePipeline
from langchain.chat_models import ChatOpenAI

# OpenAI GPT模型
llm = OpenAI(
    model_name="gpt-3.5-turbo-instruct",
    temperature=0.7,
    max_tokens=1000
)

# 聊天模型
chat_model = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    temperature=0.7
)

# Hugging Face模型
hf_llm = HuggingFacePipeline.from_model_id(
    model_id="microsoft/DialoGPT-medium",
    task="text-generation"
)

4.1.2 模型配置与优化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
from langchain.llms import OpenAI
from langchain.callbacks import get_openai_callback

# 配置模型参数
llm = OpenAI(
    temperature=0.7,        # 控制输出随机性
    max_tokens=1000,        # 最大输出长度
    top_p=0.9,             # 核采样参数
    frequency_penalty=0.1,  # 频率惩罚
    presence_penalty=0.1    # 存在惩罚
)

# 监控API调用成本
with get_openai_callback() as cb:
    result = llm("解释什么是机器学习")
    print(f"总计tokens: {cb.total_tokens}")
    print(f"总计成本: ${cb.total_cost}")

4.2 Prompts（提示）

Prompts组件负责管理和优化提示模板，是与LLM交互的关键接口。

4.2.1 提示模板

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)

# 基础提示模板
prompt_template = PromptTemplate(
    input_variables=["product", "audience"],
    template="""
    为{product}写一份针对{audience}的营销文案。
    要求：
    1. 突出产品特点
    2. 符合目标受众需求
    3. 语言生动有趣
    
    营销文案：
    """
)

# 聊天提示模板
system_template = "你是一个专业的{role}，请根据用户需求提供专业建议。"
system_message_prompt = SystemMessagePromptTemplate.from_template(system_template)

human_template = "我需要关于{topic}的建议，具体情况是：{context}"
human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)

chat_prompt = ChatPromptTemplate.from_messages([
    system_message_prompt,
    human_message_prompt
])

4.2.2 Few-shot提示

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from langchain.prompts import FewShotPromptTemplate

# 定义示例
examples = [
    {
        "question": "什么是Python？",
        "answer": "Python是一种高级编程语言，以其简洁的语法和强大的功能而闻名。"
    },
    {
        "question": "什么是机器学习？",
        "answer": "机器学习是人工智能的一个分支，通过算法让计算机从数据中学习模式。"
    }
]

# 示例模板
example_prompt = PromptTemplate(
    input_variables=["question", "answer"],
    template="问题：{question}\n答案：{answer}"
)

# Few-shot提示模板
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="以下是一些问答示例：",
    suffix="问题：{input}\n答案：",
    input_variables=["input"]
)

4.3 Chains（链）

Chains是LangChain的核心概念，允许将多个组件链接起来形成复杂的工作流。

4.3.1 基础链类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langchain.chains import LLMChain, SimpleSequentialChain, SequentialChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

llm = OpenAI(temperature=0.7)

# LLM链
prompt = PromptTemplate(
    input_variables=["topic"],
    template="写一篇关于{topic}的简短介绍"
)
llm_chain = LLMChain(llm=llm, prompt=prompt)

# 顺序链
first_prompt = PromptTemplate(
    input_variables=["topic"],
    template="为{topic}生成一个创意标题"
)
first_chain = LLMChain(llm=llm, prompt=first_prompt)

second_prompt = PromptTemplate(
    input_variables=["title"],
    template="基于标题'{title}'写一段开头"
)
second_chain = LLMChain(llm=llm, prompt=second_prompt)

sequential_chain = SimpleSequentialChain(
    chains=[first_chain, second_chain],
    verbose=True
)

4.3.2 复杂链组合

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from langchain.chains import SequentialChain

# 多输入输出的顺序链
title_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["topic"],
        template="为{topic}生成一个吸引人的标题"
    ),
    output_key="title"
)

content_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["topic", "title"],
        template="基于主题'{topic}'和标题'{title}'，写一篇详细的文章"
    ),
    output_key="content"
)

summary_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate(
        input_variables=["content"],
        template="为以下文章写一个简洁的摘要：\n{content}"
    ),
    output_key="summary"
)

overall_chain = SequentialChain(
    chains=[title_chain, content_chain, summary_chain],
    input_variables=["topic"],
    output_variables=["title", "content", "summary"],
    verbose=True
)

4.4 Agents（代理）

Agents是LangChain中最强大的组件之一，能够根据用户输入自主决策使用哪些工具。

4.4.1 代理类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langchain.agents import initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun, Tool
from langchain.llms import OpenAI
import requests

llm = OpenAI(temperature=0)

# 定义工具
search = DuckDuckGoSearchRun()

def get_weather(city: str) -> str:
    """获取指定城市的天气信息"""
    # 这里应该调用真实的天气API
    return f"{city}的天气：晴天，温度25°C"

weather_tool = Tool(
    name="Weather",
    description="获取城市天气信息",
    func=get_weather
)

tools = [search, weather_tool]

# 初始化代理
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

4.4.2 自定义代理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.agents import Tool, AgentExecutor, BaseMultiActionAgent
from langchain.schema import AgentAction, AgentFinish
from typing import List, Union

class CustomAgent(BaseMultiActionAgent):
    """自定义代理实现"""
    
    def plan(self, intermediate_steps, **kwargs) -> Union[List[AgentAction], AgentFinish]:
        """规划下一步行动"""
        user_input = kwargs.get("input", "")
        
        if "天气" in user_input:
            return [AgentAction(
                tool="Weather",
                tool_input=user_input,
                log="用户询问天气，使用天气工具"
            )]
        elif "搜索" in user_input:
            return [AgentAction(
                tool="Search",
                tool_input=user_input,
                log="用户需要搜索，使用搜索工具"
            )]
        else:
            return AgentFinish(
                return_values={"output": "我不确定如何处理这个请求"},
                log="无法确定合适的工具"
            )
    
    @property
    def input_keys(self):
        return ["input"]

4.5 Memory（记忆）

Memory组件负责管理对话历史和上下文信息，使AI应用能够保持连贯的对话。

4.5.1 记忆类型

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    ConversationSummaryBufferMemory
)
from langchain.chains import ConversationChain

llm = OpenAI(temperature=0)

# 缓冲记忆 - 保存所有对话历史
buffer_memory = ConversationBufferMemory()

# 窗口记忆 - 只保存最近k轮对话
window_memory = ConversationBufferWindowMemory(k=5)

# 摘要记忆 - 对历史对话进行摘要
summary_memory = ConversationSummaryMemory(llm=llm)

# 摘要缓冲记忆 - 结合摘要和缓冲
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=100
)

# 使用记忆的对话链
conversation = ConversationChain(
    llm=llm,
    memory=buffer_memory,
    verbose=True
)

4.5.2 自定义记忆

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langchain.memory.chat_memory import BaseChatMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage

class CustomMemory(BaseChatMemory):
    """自定义记忆实现"""
    
    def __init__(self, max_messages: int = 10):
        super().__init__()
        self.max_messages = max_messages
        self.messages = []
    
    def save_context(self, inputs: dict, outputs: dict):
        """保存对话上下文"""
        human_message = HumanMessage(content=inputs.get("input", ""))
        ai_message = AIMessage(content=outputs.get("output", ""))
        
        self.messages.extend([human_message, ai_message])
        
        # 保持消息数量在限制内
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]
    
    def clear(self):
        """清空记忆"""
        self.messages = []

4.6 Document Loaders（文档加载器）

Document Loaders负责从各种数据源加载和处理文档。

4.6.1 常用文档加载器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.document_loaders import (
    TextLoader,
    PyPDFLoader,
    CSVLoader,
    UnstructuredHTMLLoader,
    DirectoryLoader,
    WebBaseLoader
)

# 文本文件加载器
text_loader = TextLoader("document.txt")
text_docs = text_loader.load()

# PDF加载器
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()

# CSV加载器
csv_loader = CSVLoader("data.csv")
csv_docs = csv_loader.load()

# 网页加载器
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()

# 目录加载器
directory_loader = DirectoryLoader(
    "documents/",
    glob="**/*.txt",
    loader_cls=TextLoader
)
directory_docs = directory_loader.load()

4.6.2 自定义文档加载器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from langchain.document_loaders.base import BaseLoader
from langchain.schema import Document
import json

class JSONLoader(BaseLoader):
    """自定义JSON文档加载器"""
    
    def __init__(self, file_path: str, content_key: str = "content"):
        self.file_path = file_path
        self.content_key = content_key
    
    def load(self) -> List[Document]:
        """加载JSON文档"""
        documents = []
        
        with open(self.file_path, 'r', encoding='utf-8') as file:
            data = json.load(file)
            
            if isinstance(data, list):
                for item in data:
                    content = item.get(self.content_key, "")
                    metadata = {k: v for k, v in item.items() if k != self.content_key}
                    documents.append(Document(page_content=content, metadata=metadata))
            else:
                content = data.get(self.content_key, "")
                metadata = {k: v for k, v in data.items() if k != self.content_key}
                documents.append(Document(page_content=content, metadata=metadata))
        
        return documents

4.7 Vector Stores（向量存储）

Vector Stores用于存储和检索文档的向量表示，是实现语义搜索的关键组件。

4.7.1 向量存储实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.vectorstores import Chroma, FAISS, Pinecone
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.text_splitter import CharacterTextSplitter

# 初始化嵌入模型
embeddings = OpenAIEmbeddings()
# 或使用开源嵌入模型
# embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")

# 文档分割
text_splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
texts = text_splitter.split_documents(documents)

# Chroma向量存储
chroma_db = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# FAISS向量存储
faiss_db = FAISS.from_documents(
    documents=texts,
    embedding=embeddings
)

# 保存和加载FAISS索引
faiss_db.save_local("faiss_index")
loaded_faiss_db = FAISS.load_local("faiss_index", embeddings)

4.7.2 向量检索

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# 相似性搜索
query = "什么是机器学习？"
similar_docs = chroma_db.similarity_search(query, k=3)

# 带分数的相似性搜索
similar_docs_with_scores = chroma_db.similarity_search_with_score(query, k=3)

# 最大边际相关性搜索
mmr_docs = chroma_db.max_marginal_relevance_search(query, k=3)

# 自定义过滤器
filtered_docs = chroma_db.similarity_search(
    query,
    k=3,
    filter={"source": "document.pdf"}
)

4.8 Retrievers（检索器）

Retrievers提供了更高级的检索接口，支持多种检索策略。

4.8.1 基础检索器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from langchain.retrievers import (
    VectorStoreRetriever,
    BM25Retriever,
    EnsembleRetriever
)

# 向量存储检索器
vector_retriever = VectorStoreRetriever(
    vectorstore=chroma_db,
    search_kwargs={"k": 5}
)

# BM25检索器（基于关键词）
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# 集成检索器（结合多种检索方法）
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]
)

4.8.2 自定义检索器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langchain.retrievers.base import BaseRetriever
from langchain.schema import Document
from typing import List

class CustomRetriever(BaseRetriever):
    """自定义检索器"""
    
    def __init__(self, documents: List[Document]):
        self.documents = documents
    
    def get_relevant_documents(self, query: str) -> List[Document]:
        """获取相关文档"""
        # 实现自定义检索逻辑
        relevant_docs = []
        query_lower = query.lower()
        
        for doc in self.documents:
            if query_lower in doc.page_content.lower():
                relevant_docs.append(doc)
        
        return relevant_docs[:5]  # 返回前5个相关文档
    
    async def aget_relevant_documents(self, query: str) -> List[Document]:
        """异步获取相关文档"""
        return self.get_relevant_documents(query)

5. 实际应用场景

5.1 问答系统

构建基于文档的智能问答系统：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 构建问答系统
def build_qa_system(documents):
    # 创建向量存储
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_documents(documents, embeddings)
    
    # 创建检索器
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    
    # 创建问答链
    qa_chain = RetrievalQA.from_chain_type(
        llm=OpenAI(temperature=0),
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )
    
    return qa_chain

# 使用问答系统
qa_system = build_qa_system(documents)
result = qa_system({"query": "什么是深度学习？"})
print(f"答案: {result['result']}")
print(f"来源: {result['source_documents']}")

5.2 文档分析

实现智能文档分析和摘要：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain.chains.summarize import load_summarize_chain
from langchain.chains import AnalyzeDocumentChain

def analyze_document(document_path):
    # 加载文档
    loader = PyPDFLoader(document_path)
    documents = loader.load()
    
    # 创建摘要链
    summarize_chain = load_summarize_chain(
        llm=OpenAI(temperature=0),
        chain_type="map_reduce"
    )
    
    # 创建文档分析链
    analyze_chain = AnalyzeDocumentChain(
        combine_docs_chain=summarize_chain
    )
    
    # 分析文档
    summary = analyze_chain.run(documents)
    return summary

# 使用文档分析
summary = analyze_document("research_paper.pdf")
print(f"文档摘要: {summary}")

5.3 代码生成

构建智能代码生成助手：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

def create_code_generator():
    prompt_template = PromptTemplate(
        input_variables=["language", "description", "requirements"],
        template="""
        请用{language}编写代码来实现以下功能：
        
        功能描述：{description}
        
        具体要求：
        {requirements}
        
        请提供完整的、可运行的代码，并包含必要的注释。
        
        代码：
        ```{language}
        """
    )
    
    code_chain = LLMChain(
        llm=OpenAI(temperature=0.3),
        prompt=prompt_template
    )
    
    return code_chain

# 使用代码生成器
code_generator = create_code_generator()
result = code_generator.run({
    "language": "Python",
    "description": "实现一个简单的计算器",
    "requirements": "支持加减乘除运算，包含错误处理"
})
print(result)

5.4 智能助手

构建多功能智能助手：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
import requests
import json

def create_smart_assistant():
    # 定义工具
    def search_web(query: str) -> str:
        """搜索网络信息"""
        # 实现网络搜索逻辑
        return f"搜索结果：{query}"
    
    def calculate(expression: str) -> str:
        """计算数学表达式"""
        try:
            result = eval(expression)
            return f"计算结果：{result}"
        except:
            return "计算错误，请检查表达式"
    
    def get_weather(city: str) -> str:
        """获取天气信息"""
        # 实现天气查询逻辑
        return f"{city}的天气：晴天，25°C"
    
    tools = [
        Tool(name="Search", description="搜索网络信息", func=search_web),
        Tool(name="Calculator", description="计算数学表达式", func=calculate),
        Tool(name="Weather", description="获取天气信息", func=get_weather)
    ]
    
    # 创建代理
    assistant = initialize_agent(
        tools=tools,
        llm=OpenAI(temperature=0),
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True
    )
    
    return assistant

# 使用智能助手
assistant = create_smart_assistant()
response = assistant.run("北京的天气怎么样？")
print(response)

6. 开发实践指南

6.1 环境搭建

6.1.1 安装依赖

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
# 基础安装
pip install langchain

# 安装特定集成
pip install langchain[openai]  # OpenAI集成
pip install langchain[anthropic]  # Anthropic集成
pip install langchain[huggingface]  # Hugging Face集成

# 安装向量存储依赖
pip install chromadb  # Chroma
pip install faiss-cpu  # FAISS
pip install pinecone-client  # Pinecone

# 安装文档处理依赖
pip install pypdf  # PDF处理
pip install unstructured  # 非结构化文档
pip install beautifulsoup4  # HTML处理

6.1.2 环境配置

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 设置API密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key"
os.environ["HUGGINGFACEHUB_API_TOKEN"] = "your-huggingface-token"

# 设置代理（如果需要）
os.environ["HTTP_PROXY"] = "http://proxy.example.com:8080"
os.environ["HTTPS_PROXY"] = "https://proxy.example.com:8080"

6.2 基础使用示例

6.2.1 简单的LLM调用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from langchain.llms import OpenAI

# 初始化模型
llm = OpenAI(temperature=0.7)

# 简单调用
response = llm("解释什么是人工智能")
print(response)

# 批量调用
questions = [
    "什么是机器学习？",
    "什么是深度学习？",
    "什么是神经网络？"
]
responses = llm.generate(questions)
for response in responses.generations:
    print(response[0].text)

6.2.2 使用提示模板

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# 创建提示模板
prompt = PromptTemplate(
    input_variables=["topic", "audience"],
    template="为{audience}解释{topic}，要求通俗易懂"
)

# 创建链
chain = LLMChain(llm=llm, prompt=prompt)

# 执行链
result = chain.run(topic="量子计算", audience="中学生")
print(result)

6.3 高级应用开发

6.3.1 构建RAG系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

def build_rag_system(data_directory: str):
    # 1. 加载文档
    loader = DirectoryLoader(data_directory, glob="**/*.txt")
    documents = loader.load()
    
    # 2. 分割文档
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )
    texts = text_splitter.split_documents(documents)
    
    # 3. 创建向量存储
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_documents(texts, embeddings)
    
    # 4. 创建检索器
    retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}
    )
    
    # 5. 创建QA链
    qa_chain = RetrievalQA.from_chain_type(
        llm=OpenAI(temperature=0),
        chain_type="stuff",
        retriever=retriever,
        return_source_documents=True
    )
    
    return qa_chain

# 使用RAG系统
rag_system = build_rag_system("./documents")
result = rag_system({"query": "如何优化机器学习模型？"})

6.3.2 构建对话系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate

def build_conversation_system():
    # 自定义对话提示
    template = """
    你是一个友好的AI助手，请根据对话历史和用户输入提供有帮助的回答。
    
    对话历史:
    {history}
    
    用户: {input}
    助手:"""
    
    prompt = PromptTemplate(
        input_variables=["history", "input"],
        template=template
    )
    
    # 创建记忆
    memory = ConversationBufferWindowMemory(k=5)
    
    # 创建对话链
    conversation = ConversationChain(
        llm=OpenAI(temperature=0.7),
        prompt=prompt,
        memory=memory,
        verbose=True
    )
    
    return conversation

# 使用对话系统
conversation_system = build_conversation_system()

# 多轮对话
print(conversation_system.predict(input="你好，我想了解机器学习"))
print(conversation_system.predict(input="能给我推荐一些学习资源吗？"))
print(conversation_system.predict(input="我应该从哪里开始学习？"))

7. 最佳实践

7.1 性能优化

7.1.1 缓存策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache
import langchain

# 内存缓存
set_llm_cache(InMemoryCache())

# SQLite缓存
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

# 使用缓存
llm = OpenAI(temperature=0)
result1 = llm("什么是机器学习？")  # 第一次调用API
result2 = llm("什么是机器学习？")  # 从缓存获取

7.1.2 批处理优化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from langchain.llms import OpenAI
from langchain.schema import LLMResult

# 批量处理
llm = OpenAI(temperature=0)
prompts = [
    "解释机器学习",
    "解释深度学习",
    "解释神经网络"
]

# 批量调用比单独调用更高效
batch_result = llm.generate(prompts)

7.1.3 异步处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import asyncio
from langchain.llms import OpenAI

async def async_llm_call():
    llm = OpenAI(temperature=0)
    
    # 异步调用
    tasks = [
        llm.agenerate(["什么是AI？"]),
        llm.agenerate(["什么是ML？"]),
        llm.agenerate(["什么是DL？"])
    ]
    
    results = await asyncio.gather(*tasks)
    return results

# 运行异步任务
results = asyncio.run(async_llm_call())

7.2 错误处理

7.2.1 重试机制

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langchain.llms import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
import logging

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def robust_llm_call(prompt: str):
    try:
        llm = OpenAI(temperature=0)
        return llm(prompt)
    except Exception as e:
        logging.error(f"LLM调用失败: {e}")
        raise

# 使用重试机制
try:
    result = robust_llm_call("解释量子计算")
    print(result)
except Exception as e:
    print(f"最终失败: {e}")

7.2.2 异常处理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from langchain.llms import OpenAI
from langchain.schema import OutputParserException

def safe_llm_chain(prompt: str):
    try:
        llm = OpenAI(temperature=0)
        result = llm(prompt)
        return {"success": True, "result": result}
    except OutputParserException as e:
        return {"success": False, "error": f"输出解析错误: {e}"}
    except Exception as e:
        return {"success": False, "error": f"未知错误: {e}"}

# 安全调用
response = safe_llm_chain("解释机器学习")
if response["success"]:
    print(response["result"])
else:
    print(f"错误: {response['error']}")

7.3 安全考虑

7.3.1 输入验证

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import re
from typing import Optional

def validate_input(user_input: str) -> Optional[str]:
    """验证用户输入"""
    # 检查输入长度
    if len(user_input) > 1000:
        return "输入过长"
    
    # 检查恶意内容
    malicious_patterns = [
        r"<script.*?>.*?</script>",
        r"javascript:",
        r"on\w+\s*=",
    ]
    
    for pattern in malicious_patterns:
        if re.search(pattern, user_input, re.IGNORECASE):
            return "检测到潜在恶意内容"
    
    return None

def safe_process_input(user_input: str):
    # 验证输入
    error = validate_input(user_input)
    if error:
        return f"输入验证失败: {error}"
    
    # 处理输入
    llm = OpenAI(temperature=0)
    return llm(user_input)

7.3.2 API密钥管理

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import os
from cryptography.fernet import Fernet

class SecureConfig:
    """安全配置管理"""
    
    def __init__(self):
        self.key = os.environ.get("ENCRYPTION_KEY")
        if not self.key:
            self.key = Fernet.generate_key()
        self.cipher = Fernet(self.key)
    
    def encrypt_api_key(self, api_key: str) -> str:
        """加密API密钥"""
        return self.cipher.encrypt(api_key.encode()).decode()
    
    def decrypt_api_key(self, encrypted_key: str) -> str:
        """解密API密钥"""
        return self.cipher.decrypt(encrypted_key.encode()).decode()

# 使用安全配置
config = SecureConfig()
encrypted_key = config.encrypt_api_key("your-api-key")
decrypted_key = config.decrypt_api_key(encrypted_key)

8. 生态系统与集成

8.1 第三方集成

8.1.1 数据库集成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from langchain.vectorstores import Pinecone, Weaviate
from langchain.document_loaders import MongoDBLoader
import pinecone

# Pinecone集成
pinecone.init(
    api_key="your-pinecone-api-key",
    environment="your-environment"
)

pinecone_store = Pinecone.from_documents(
    documents,
    embeddings,
    index_name="langchain-index"
)

# MongoDB集成
mongo_loader = MongoDBLoader(
    connection_string="mongodb://localhost:27017/",
    db_name="your_db",
    collection_name="your_collection"
)
mongo_docs = mongo_loader.load()

8.1.2 Web框架集成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from flask import Flask, request, jsonify
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

app = Flask(__name__)

# 初始化对话系统
memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=OpenAI(temperature=0.7),
    memory=memory
)

@app.route('/chat', methods=['POST'])
def chat():
    user_input = request.json.get('message')
    response = conversation.predict(input=user_input)
    return jsonify({'response': response})

if __name__ == '__main__':
    app.run(debug=True)

8.2 社区资源

8.2.1 有用的工具和库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
# LangSmith - 调试和监控
from langsmith import Client

client = Client()

# LangServe - 部署LangChain应用
from langserve import add_routes
from fastapi import FastAPI

app = FastAPI()
add_routes(app, conversation, path="/chat")

# LangGraph - 构建复杂的工作流
from langgraph import Graph

graph = Graph()
graph.add_node("start", lambda x: x)
graph.add_node("process", lambda x: process_data(x))
graph.add_edge("start", "process")

9. 未来发展趋势

9.1 应用场景

mindmap root((LangChain未来应用)) 企业级应用智能客服文档管理知识库问答业务流程自动化教育领域个性化学习智能辅导自动评分课程推荐创意产业内容生成创意写作设计辅助音乐创作科研领域文献分析实验设计数据解释假设生成

1. 引言#

2. LangChain概述#

2.1 什么是LangChain#

2.2 核心设计理念#

2.3 主要特性#

3. LangChain架构详解#

3.1 整体架构#

3.2 核心组件#

4. 核心组件深入分析#

4.1 Models（模型）#

4.1.1 LLM（大语言模型）#

4.1.2 模型配置与优化#

4.2 Prompts（提示）#

4.2.1 提示模板#

4.2.2 Few-shot提示#

4.3 Chains（链）#

4.3.1 基础链类型#

4.3.2 复杂链组合#

4.4 Agents（代理）#

4.4.1 代理类型#

4.4.2 自定义代理#

4.5 Memory（记忆）#

4.5.1 记忆类型#

4.5.2 自定义记忆#

4.6 Document Loaders（文档加载器）#

4.6.1 常用文档加载器#

4.6.2 自定义文档加载器#

4.7 Vector Stores（向量存储）#

4.7.1 向量存储实现#

4.7.2 向量检索#

4.8 Retrievers（检索器）#

4.8.1 基础检索器#

4.8.2 自定义检索器#

5. 实际应用场景#

5.1 问答系统#

5.2 文档分析#

5.3 代码生成#

5.4 智能助手#

6. 开发实践指南#

6.1 环境搭建#

6.1.1 安装依赖#

6.1.2 环境配置#

6.2 基础使用示例#

6.2.1 简单的LLM调用#

6.2.2 使用提示模板#

6.3 高级应用开发#

6.3.1 构建RAG系统#

6.3.2 构建对话系统#

7. 最佳实践#

7.1 性能优化#

7.1.1 缓存策略#

7.1.2 批处理优化#

7.1.3 异步处理#

7.2 错误处理#

7.2.1 重试机制#

7.2.2 异常处理#

7.3 安全考虑#

7.3.1 输入验证#

7.3.2 API密钥管理#

8. 生态系统与集成#

8.1 第三方集成#

8.1.1 数据库集成#

8.1.2 Web框架集成#

8.2 社区资源#

8.2.1 有用的工具和库#

9. 未来发展趋势#

9.1 应用场景#