1.3 基础组件概览

1.3 基础组件概览

学习目标

读完本节，你将能够：

• 全面了解LangChain的核心组件体系
• 掌握各个组件的功能和适用场景
• 学会组件的正确组合方式
• 建立组件间的协作关系理解
• 为后续深入学习奠定基础

1.3.1 组件架构总览

🏗️ LangChain组件层级

LangChain的组件设计采用了清晰的层次结构，每个组件都有特定的职责和定位：

🔑 组件设计原则

1. 模块化设计

每个组件都是独立的模块，可以单独测试和替换：

# 独立组件示例
from langchain_openai import ChatOpenAI  # LLM组件
from langchain.prompts import ChatPromptTemplate  # 提示词组件
from langchain.memory import ConversationBufferMemory  # 记忆组件
from langchain.vectorstores import Chroma  # 向量存储组件
​

2. 统一接口

同类组件使用统一的接口设计：

# LLM统一接口
class BaseLLM:
    def invoke(self, prompt: str) -> str:
        pass
    
    def batch_invoke(self, prompts: List[str]) -> List[str]:
        pass

# 具体实现
class OpenAIChat(BaseLLM):
    def invoke(self, prompt: str) -> str:
        # OpenAI具体实现
        pass

class AnthropicChat(BaseLLM):
    def invoke(self, prompt: str) -> str:
        # Anthropic具体实现
        pass
​

3. 可组合性

组件之间可以灵活组合：

# 组件组合示例
from langchain.chains import LLMChain
from langchain.agents import AgentExecutor

# 组合1：链式调用
analysis_chain = LLMChain(llm=llm, prompt=prompt_template)

# 组合2：智能体
agent = AgentExecutor(
    agent=react_agent,
    tools=[search_tool, calculator_tool],
    llm=llm
)
​

1.3.2 LLM接口组件

🤖 LLM接口概述

LangChain提供了统一的LLM接口，支持多种大语言模型：

核心接口设计

from langchain.llms.base import BaseLLM
from langchain.chat_models.base import BaseChatModel

class BaseLLM:
    """基础LLM接口"""
    def invoke(self, prompt: str) -> str:
        """同步调用"""
        pass
    
    def generate(self, prompts: List[str]) -> List[str]:
        """批量生成"""
        pass

class BaseChatModel(BaseLLM):
    """聊天模型接口（继承自LLM）"""
    def invoke(self, messages: List[dict]) -> dict:
        """消息列表调用"""
        pass
​

支持的模型提供商

🔧 OpenAI集成

基础使用

from langchain_openai import ChatOpenAI

# 基础配置
llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.7,
    max_tokens=2000,
    top_p=0.9,
    frequency_penalty=0.0,
    presence_penalty=0.0
)

# 调用示例
response = llm.invoke("你好，请介绍一下自己")
print(response.content)
​

高级配置

from langchain_openai import ChatOpenAI

# 高级配置
advanced_llm = ChatOpenAI(
    model="gpt-4-1106-preview",  # 最新模型
    temperature=0.3,              # 低温度，更确定性
    max_tokens=4000,            # 更长输出
    top_p=0.8,                  # 核心采样
    frequency_penalty=0.1,       # 减少重复
    presence_penalty=0.2,       # 鼓励多样性
    request_timeout=60,         # 请求超时
    retry_attempts=3,          # 重试次数
    retry_delay=5,            # 重试延迟
)

# 流式输出
for chunk in advanced_llm.stream("请详细解释量子计算"):
    print(chunk.content, end="", flush=True)
​

批量处理

from langchain_openai import ChatOpenAI

llm = ChatOpenAI()

# 批量调用
prompts = [
    "解释什么是机器学习",
    "深度学习与机器学习的区别",
    "神经网络的基本原理"
]

responses = llm.generate(prompts)
for i, response in enumerate(responses.generations):
    print(f"问题 {i+1}: {response[0].text}")
​

🔧 Anthropic集成

Claude模型配置

from langchain_anthropic import ChatAnthropic

# Claude 3 Sonnet
claude_sonnet = ChatAnthropic(
    model="claude-3-sonnet-20240229",
    temperature=0.5,
    max_tokens=1000
)

# Claude 3 Opus（更强能力）
claude_opus = ChatAnthropic(
    model="claude-3-opus-20240229",
    temperature=0.3,
    max_tokens=2000
)

# Claude 3 Haiku（快速响应）
claude_haiku = ChatAnthropic(
    model="claude-3-haiku-20240307",
    temperature=0.7,
    max_tokens=500
)
​

系统消息配置

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage

claude = ChatAnthropic()

# 使用系统消息
system_prompt = "你是一个专业的数据科学家，擅长用通俗易懂的方式解释复杂概念。"

messages = [
    SystemMessage(content=system_prompt),
    HumanMessage(content="请解释什么是过拟合？"),
    HumanMessage(content("那什么是欠拟合？"))
]

response = claude.invoke(messages)
print(response.content)
​

🔧 Google集成

Gemini配置

from langchain_google import ChatGoogle

# Gemini Pro
gemini_pro = ChatGoogle(
    model="gemini-pro",
    temperature=0.7,
    max_output_tokens=2048,
    top_p=0.8,
    top_k=40
)

# Gemini Ultra
gemini_ultra = ChatGoogle(
    model="gemini-ultra",
    temperature=0.3,
    max_output_tokens=4096,
)
​

多模态支持

from langchain_google import ChatGoogle
from langchain_core.messages import HumanMessage
from langchain_core.documents import Document

# 支持文本和图像
gemini = ChatGoogle()

# 纯文本
text_message = HumanMessage(content="解释这个机器学习算法")
text_response = gemini.invoke([text_message])

# 图像+文本（需要实际图像数据）
# image_message = HumanMessage(
#     content=[
#         {"type": "text", "text": "描述这张图片"},
#         {"type": "image_url", "image_url": "data:image/jpeg;base64,..."}
#     ]
# )
# image_response = gemini.invoke([image_message])
​

1.3.3 提示词组件

📝 提示词模板系统

LangChain提供了强大的提示词模板系统，支持动态提示词生成。

基础模板

from langchain.prompts import ChatPromptTemplate

# 简单字符串模板
simple_template = ChatPromptTemplate.from_template(
    "你是一个{role}专家，请针对{topic}提供专业建议。"
)

# 使用模板
prompt = simple_template.format(
    role="数据科学",
    topic="机器学习模型选择"
)

print(prompt)
# 输出: 你是一个数据科学专家，请针对机器学习模型选择提供专业建议。
​

消息模板

from langchain.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage

# 消息模板
message_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{personality}的助手"),
    ("human", "{user_input}"),
    ("assistant", "我先理解你的问题..."),
    ("human", "请{task}"),
])

# 使用消息模板
messages = message_template.format_messages(
    personality="友好且专业",
    user_input="我想了解深度学习",
    task="详细解释什么是卷积神经网络"
)

# 显示消息格式
for msg in messages:
    print(f"{msg.type}: {msg.content}")
​

F-string模板

from langchain.prompts import PromptTemplate

# f-string风格模板
fstring_template = PromptTemplate(
    input_variables=["topic", "difficulty"],
    template=f"""
    你是一位{difficulty}的{topic}导师。
    
    请按照以下步骤指导学习：
    1. 介绍基础概念
    2. 提供实际案例
    3. 练习建议
    4. 进阶资源
    
    主题: {topic}
    难度级别: {difficulty}
    """
)

# 使用f-string模板
prompt = fstring_template.format(
    topic="Python编程",
    difficulty="中级"
)
print(prompt)
​

自定义模板类

from langchain.prompts import BasePromptTemplate
from typing import List, Dict, Any

class CustomPromptTemplate(BasePromptTemplate):
    """自定义提示词模板"""
    
    def __init__(self, template: str, input_variables: List[str]):
        super().__init__(input_variables=input_variables, template=template)
    
    def format(self, **kwargs) -> str:
        """自定义格式化逻辑"""
        # 添加元数据
        metadata = {
            "timestamp": kwargs.get("timestamp", ""),
            "user_id": kwargs.get("user_id", ""),
            "session_id": kwargs.get("session_id", "")
        }
        
        # 格式化基础模板
        formatted = super().format(**kwargs)
        
        # 添加元数据前缀
        metadata_str = "\n".join([f"[{k}]: {v}" for k, v in metadata.items() if v])
        
        return f"{metadata_str}\n\n{formatted}"

# 使用自定义模板
custom_template = CustomPromptTemplate(
    template="请根据以下信息回答问题：\n{context}\n\n用户问题：{question}",
    input_variables=["context", "question", "timestamp", "user_id"]
)

prompt = custom_template.format(
    context="机器学习是人工智能的一个分支...",
    question="什么是监督学习？",
    timestamp="2024-01-01",
    user_id="user123"
)
​

🎨 提示词优化

动态参数调整

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 动态模板
dynamic_template = ChatPromptTemplate.from_template(
    """
    你是一个{expertise}领域的专家，当前分析场景是：{scenario}
    
    针对用户的问题：{user_question}
    
    请提供以下内容：
    1. 核心概念解释
    2. 实际应用案例
    3. 常见问题解答
    4. 学习资源推荐
    
    回答要求：
    - 深度：{depth_level}
    - 语言：{language}
    - 格式：{format_style}
    """
)

# 根据用户需求动态调整
def generate_prompt(user_question: str, scenario: str, expertise: str = "AI"):
    """生成个性化提示词"""
    
    # 根据问题复杂度调整深度
    depth_map = {
        "简单": "基础",
        "中等": "中等", 
        "复杂": "深入"
    }
    
    # 根据用户偏好调整
    prompt = dynamic_template.format(
        expertise=expertise,
        scenario=scenario,
        user_question=user_question,
        depth_level=depth_map.get("中等", "中等"),
        language="中文",
        format_style="结构化列表"
    )
    
    return prompt

# 使用示例
prompt = generate_prompt(
    "如何开始学习机器学习？",
    "初学者入门场景",
    expertise="数据科学"
)
​

提示词链

from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 预处理链
preprocess_prompt = ChatPromptTemplate.from_template(
    "将以下问题转换为标准格式：{question}"
)

# 主分析链
main_prompt = ChatPromptTemplate.from_template(
    "根据预处理后的回答进行详细分析：{processed_question}"
)

# 创建链式处理
def process_complex_question(question: str):
    """复杂问题处理链"""
    
    # 第一阶段：问题预处理
    preprocess_chain = LLMChain(
        llm=llm,
        prompt=preprocess_prompt
    )
    
    processed = preprocess_chain.run(question=question)
    
    # 第二阶段：详细分析
    main_chain = LLMChain(
        llm=llm,
        prompt=main_prompt
    )
    
    analysis = main_chain.run(processed_question=processed)
    
    return {
        "original_question": question,
        "processed_question": processed,
        "analysis": analysis
    }
​

1.3.4 文档处理组件

📄 文档加载器

文档加载器负责从各种数据源读取并加载文档内容：

基础文本加载

from langchain.document_loaders import TextLoader

# 单个文本文件
text_loader = TextLoader("document.txt")
documents = text_loader.load()

# 批量文本文件
loaders = [
    TextLoader("doc1.txt"),
    TextLoader("doc2.txt"),
    TextLoader("doc3.txt")
]

all_documents = []
for loader in loaders:
    all_documents.extend(loader.load())
​

PDF文档加载

from langchain.document_loaders import PyPDFLoader

# PDF文件加载
pdf_loader = PyPDFLoader("report.pdf")
pdf_docs = pdf_loader.load()

# 多页PDF处理
def process_pdf(file_path: str, chunk_size: int = 1000):
    """处理大型PDF文件"""
    loader = PyPDFLoader(file_path)
    documents = loader.load()
    
    # 分块处理
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=200
    )
    
    chunks = splitter.split_documents(documents)
    return chunks
​

网页内容加载

from langchain.document_loaders import WebBaseLoader

# 单个网页
web_loader = WebBaseLoader(["https://example.com"])
web_docs = web_loader.load()

# 多个网页
urls = [
    "https://site1.com/article1",
    "https://site2.com/blog2", 
    "https://site3.com/docs3"
]

multi_web_loader = WebBaseLoader(urls)
multi_docs = multi_web_loader.load()
​

多种文档格式支持

from langchain.document_loaders import (
    PyPDFLoader,      # PDF
    TextLoader,       # 纯文本
    CSVLoader,        # CSV文件
    UnstructuredHTMLLoader,  # HTML
    UnstructuredMarkdownLoader,  # Markdown
    JSONLoader,       # JSON文件
    NotionLoader      # Notion页面
)

# 统一接口示例
def load_documents_by_format(file_path: str, file_type: str):
    """根据文件类型选择加载器"""
    loader_map = {
        "txt": TextLoader,
        "pdf": PyPDFLoader,
        "csv": CSVLoader,
        "html": UnstructuredHTMLLoader,
        "md": UnstructuredMarkdownLoader,
        "json": JSONLoader
    }
    
    loader_class = loader_map.get(file_type)
    if not loader_class:
        raise ValueError(f"不支持的文件类型: {file_type}")
    
    loader = loader_class(file_path)
    return loader.load()
​

🗂️ 文档转换器

文档转换器用于处理和转换已加载的文档内容：

文本分割

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    TokenTextSplitter
)

# 递归字符分割器
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
)

# 字符分割器
char_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200
)

# Token分割器
token_splitter = TokenTextSplitter(
    chunk_size=200,
    chunk_overlap=50,
    encoding_name="cl100k_base"  # GPT-3.5/4的编码
)

# 分割文档
chunks = recursive_splitter.split_documents(documents)
​

文档清洗

from langchain.document_transformers import (
    Html2TextTransformer,
    CleanTransformer,
    LongContextReorder
)

# HTML转文本
html_transformer = Html2TextTransformer()
clean_texts = html_transformer.transform_documents(html_docs)

# 清理文档
clean_transformer = CleanTransformer(
    clean_whitespace=True,
    clean_headers=True,
    remove_extra_whitespace=True
)

cleaned_docs = clean_transformer.transform_documents(documents)

# 上下文重排序
reorder = LongContextReorder()
reordered_docs = reorder.transform_documents(docs)
​

🔢 向量存储组件

向量存储用于高效的语义搜索和检索：

Chroma向量存储

from langchain.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002"
)

# 从文档创建向量存储
chroma_store = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 持久化和加载
chroma_store.persist()  # 保存到磁盘
loaded_store = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)
​

FAISS向量存储

from langchain.vectorstores import FAISS

# 创建FAISS向量存储
faiss_store = FAISS.from_documents(
    documents=documents,
    embedding=embeddings
)

# 保存到本地
faiss_store.save_local("./faiss_db")

# 加载本地存储
loaded_faiss = FAISS.load_local(
    "./faiss_db",
    embeddings,
    allow_dangerous_deserialization=True
)
​

向量搜索操作

# 相似性搜索
results = chroma_store.similarity_search(
    query="机器学习算法",
    k=5
)

# 带分数的搜索
results_with_scores = chroma_store.similarity_search_with_score(
    query="深度学习",
    k=3
)

# 元数据过滤
filtered_results = chroma_store.similarity_search_with_score(
    query="算法",
    k=5,
    filter={"document_type": "research", "year": 2023}
)

# 多元搜索
mmr_results = chroma_store.max_marginal_relevance_search(
    query="机器学习",
    k=3,
    fetch_k=5,
    lambda_mult=0.5
)
​

🧠 记忆管理组件

记忆管理用于维护对话历史和上下文：

基础记忆

from langchain.memory import (
    ConversationBufferMemory,
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    ConversationSummaryBufferMemory
)

# 对话缓冲记忆
buffer_memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    human_prefix="用户",
    ai_prefix="AI助手"
)

# 滑动窗口记忆
window_memory = ConversationBufferWindowMemory(
    k=5,  # 保留最近5轮对话
    memory_key="window_history",
    return_messages=True
)

# 总结记忆
summary_memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="summary_history",
    return_messages=True
)

# 总结缓冲记忆
summary_buffer_memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=1000,
    memory_key="buffer_history",
    return_messages=True
)
​

记忆使用

# 保存对话上下文
buffer_memory.save_context(
    {"input": "你好，我想了解机器学习"},
    {"output": "你好！很高兴为你介绍机器学习的基础知识。"}
)

# 获取对话历史
chat_history = buffer_memory.load_memory_variables({})
print(chat_history["chat_history"])

# 在链中使用记忆
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

template = """
你是一个机器学习专家，请根据以下对话历史回答问题：
{chat_history}

用户问题：{question}
AI专家：
"""

prompt = PromptTemplate(
    input_variables=["chat_history", "question"],
    template=template
)

chain = LLMChain(
    llm=llm,
    prompt=prompt,
    memory=buffer_memory
)

# 使用链进行对话
response = chain.run(question="什么是监督学习？")
​

🔧 工具组件

工具组件用于与外部系统进行交互：

自定义工具

from langchain.tools import Tool
from typing import Optional

# 搜索工具
def search_web(query: str) -> str:
    """搜索网络信息"""
    import requests
    
    try:
        # 这里可以集成实际的搜索API
        response = requests.get(
            f"https://api.exam
​