LangChain 之 Models（模型）

Chat Models 聊天模型

Chat Models 聊天模型使用场景

聊天模型是 LangChain 中最常用的模型类型之一，它们专门设计用于处理对话式交互。与基础语言模型不同，聊天模型使用消息列表作为输入和输出，能够更好地理解和维持对话上下文。

主要使用场景：

1. 对话系统与聊天机器人

   • 客户服务机器人：处理客户咨询、解答常见问题
   • 个人助理：管理日程、设置提醒、回答日常问题
   • 领域专家机器人：在特定领域提供专业建议

2. 多轮交互应用

   • 教育辅导：逐步引导学生解决问题
   • 心理咨询：进行多轮对话了解用户心理状态
   • 复杂任务分解：通过多轮对话分解复杂任务

3. 代码辅助与开发

   • 代码解释：通过对话解释代码逻辑
   • 调试助手：通过交互式对话帮助定位问题
   • 技术咨询：回答开发过程中的技术问题

4. 内容创作与优化

   • 写作助手：通过多轮对话完善文章内容
   • 创意头脑风暴：通过对话激发创意想法
   • 文本优化：根据反馈逐步改进文本质量

Chat Models 聊天模型使用例子

基本使用示例

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage

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

# 创建消息
messages = [
    SystemMessage(content="你是一个专业的Python编程助手。"),
    HumanMessage(content="如何使用Python创建一个简单的Web服务器？")
]

# 调用模型
response = chat_model(messages)
print(response.content)

多轮对话示例

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 初始化模型和记忆
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)
memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=chat_model,
    memory=memory,
    verbose=True
)

# 进行多轮对话
response1 = conversation.predict(input="我想学习Python，应该从哪里开始？")
print(response1)

response2 = conversation.predict(input="你能推荐一些适合初学者的项目吗？")
print(response2)

角色扮演示例

from langchain.chat_models import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage

# 初始化模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

# 设置系统角色
system_message = SystemMessage(content="你是一位经验丰富的产品经理，擅长产品规划和市场分析。")

# 用户问题
user_message = HumanMessage(content="我有一个新的社交App想法，如何评估它的市场潜力？")

# 获取专业建议
response = chat_model([system_message, user_message])
print(response.content)

流式响应示例

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 初始化模型并设置流式回调
chat_model = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0.7
)

# 发送消息并获取流式响应
message = HumanMessage(content="请详细解释什么是机器学习，并给出实际应用案例。")
response = chat_model([message])

Chat Models 聊天模型的上下文缓存

上下文缓存是优化聊天模型性能和降低成本的重要技术。LangChain 提供了多种缓存机制来存储和重用模型响应。

1. 内存缓存 (In-Memory Cache)

from langchain.chat_models import ChatOpenAI
from langchain.cache import InMemoryCache
from langchain.schema import HumanMessage
import langchain
from langchain.globals import set_llm_cache

# 设置内存缓存
set_llm_cache(InMemoryCache())

# 初始化模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

# 第一次调用 - 会实际调用API
message = HumanMessage(content="什么是人工智能？")
response1 = chat_model([message])
print("第一次调用:", response1.content)

# 第二次调用相同问题 - 会从缓存获取
response2 = chat_model([message])
print("第二次调用:", response2.content)

2. Redis 缓存

from langchain.cache import RedisCache
from langchain.chat_models import ChatOpenAI
import redis

# 设置Redis缓存
redis_client = redis.Redis(host='localhost', port=6379, db=0)
set_llm_cache(RedisCache(redis_client=redis_client))

# 初始化模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

# 使用缓存
message = HumanMessage(content="解释量子计算的基本原理")
response = chat_model([message])
print(response.content)

3. 自定义缓存策略

from langchain.cache import BaseCache
from typing import Any, Optional
import hashlib
import json

class CustomCache(BaseCache):
    def __init__(self):
        self.cache = {}
    
    def _generate_key(self, prompt: str, llm_string: str) -> str:
        """生成缓存键"""
        combined = f"{prompt}_{llm_string}"
        return hashlib.md5(combined.encode()).hexdigest()
    
    def lookup(self, prompt: str, llm_string: str) -> Optional[Any]:
        """查找缓存"""
        key = self._generate_key(prompt, llm_string)
        return self.cache.get(key)
    
    def update(self, prompt: str, llm_string: str, response: Any) -> None:
        """更新缓存"""
        key = self._generate_key(prompt, llm_string)
        self.cache[key] = response

# 使用自定义缓存
set_llm_cache(CustomCache())
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

4. 语义缓存

语义缓存基于内容的相似性而非精确匹配来缓存和检索响应：

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.cache import BaseCache
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class SemanticCache(BaseCache):
    def __init__(self, similarity_threshold=0.9):
        self.embeddings = OpenAIEmbeddings()
        self.cache_texts = []
        self.cache_responses = []
        self.cache_embeddings = []
        self.similarity_threshold = similarity_threshold
    
    def lookup(self, prompt: str, llm_string: str) -> Optional[Any]:
        """基于语义相似性查找缓存"""
        if not self.cache_texts:
            return None
            
        # 计算当前提示的嵌入
        prompt_embedding = self.embeddings.embed_query(prompt)
        
        # 计算与缓存中所有条目的相似度
        similarities = []
        for cached_embedding in self.cache_embeddings:
            similarity = cosine_similarity(
                [prompt_embedding], 
                [cached_embedding]
            )[0][0]
            similarities.append(similarity)
        
        # 找到最相似的缓存条目
        max_similarity = max(similarities)
        if max_similarity >= self.similarity_threshold:
            best_match_index = similarities.index(max_similarity)
            return self.cache_responses[best_match_index]
        
        return None
    
    def update(self, prompt: str, llm_string: str, response: Any) -> None:
        """更新缓存"""
        self.cache_texts.append(prompt)
        self.cache_responses.append(response)
        prompt_embedding = self.embeddings.embed_query(prompt)
        self.cache_embeddings.append(prompt_embedding)

# 使用语义缓存
set_llm_cache(SemanticCache(similarity_threshold=0.85))
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7)

流式响应在LangChain中的应用

流式响应允许在模型生成完整响应之前就开始接收和处理部分输出，这对于改善用户体验和实时应用至关重要。

1. 基本流式响应

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 设置流式回调
chat_model = ChatOpenAI(
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0.7
)

# 发送消息并获取流式响应
message = HumanMessage(content="详细解释区块链技术的工作原理")
response = chat_model([message])

2. 自定义流式处理器

from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List, Optional
from langchain.schema import LLMResult

class CustomStreamingHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens = []
        self.is_streaming = False
    
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> None:
        """当LLM开始生成时调用"""
        self.is_streaming = True
        self.tokens = []
        print("开始生成响应...")
    
    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        """当生成新token时调用"""
        self.tokens.append(token)
        print(token, end='', flush=True)  # 实时打印token
    
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """当LLM完成生成时调用"""
        self.is_streaming = False
        print("\n生成完成!")
        print(f"总token数: {len(self.tokens)}")
    
    def on_llm_error(
        self, error: Exception, **kwargs: Any
    ) -> None:
        """当发生错误时调用"""
        self.is_streaming = False
        print(f"\n生成出错: {error}")

# 使用自定义流式处理器
handler = CustomStreamingHandler()
chat_model = ChatOpenAI(
    streaming=True,
    callbacks=[handler],
    temperature=0.7
)

message = HumanMessage(content="解释机器学习中的过拟合现象")
response = chat_model([message])

3. 流式响应与异步处理

import asyncio
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage

async def stream_response():
    """异步流式响应处理"""
    chat_model = ChatOpenAI(
        streaming=True,
        temperature=0.7
    )
    
    message = HumanMessage(content="详细解释深度学习中的反向传播算法")
    
    # 使用异步流式生成
    async for chunk in chat_model.astream([message]):
        print(chunk.content, end='', flush=True)
        await asyncio.sleep(0.01)  # 模拟处理延迟

# 运行异步流式响应
asyncio.run(stream_response())

4. 流式响应与实时UI更新

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
import time

class StreamingUIUpdater:
    def __init__(self):
        self.accumulated_text = ""
        self.update_interval = 0.1  # UI更新间隔(秒)
        self.last_update_time = 0
    
    def update_ui(self, token: str):
        """模拟UI更新"""
        self.accumulated_text += token
        current_time = time.time()
        
        # 控制UI更新频率
        if current_time - self.last_update_time >= self.update_interval:
            print(f"\r[UI更新] 当前长度: {len(self.accumulated_text)} 字符", end='', flush=True)
            self.last_update_time = current_time
    
    def on_token(self, token: str):
        """处理每个token"""
        self.update_ui(token)
    
    def on_complete(self):
        """完成时的处理"""
        print(f"\n\n完整响应长度: {len(self.accumulated_text)} 字符")

# 创建流式处理器
ui_updater = StreamingUIUpdater()

# 创建自定义回调处理器
class UIUpdateHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        ui_updater.on_token(token)
    
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        ui_updater.on_complete()

# 使用流式响应更新UI
chat_model = ChatOpenAI(
    streaming=True,
    callbacks=[UIUpdateHandler()],
    temperature=0.7
)

message = HumanMessage(content="解释云计算的三种服务模型")
response = chat_model([message])

5. 流式响应的高级应用

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage
from langchain.callbacks.base import BaseCallbackHandler
import json
import re

class StructuredStreamingHandler(BaseCallbackHandler):
    """处理结构化数据的流式响应"""
    
    def __init__(self):
        self.buffer = ""
        self.structured_data = []
        self.current_object = {}
    
    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
        """处理每个token并尝试解析结构化数据"""
        self.buffer += token
        
        # 尝试解析JSON对象
        json_objects = self.extract_json_objects(self.buffer)
        
        for obj in json_objects:
            if obj not in self.structured_data:
                self.structured_data.append(obj)
                print(f"解析到结构化数据: {json.dumps(obj, ensure_ascii=False)}")
    
    def extract_json_objects(self, text: str):
        """从文本中提取JSON对象"""
        json_objects = []
        # 简单的JSON对象匹配正则
        pattern = r'\{[^{}]*\}'
        matches = re.findall(pattern, text)
        
        for match in matches:
            try:
                obj = json.loads(match)
                json_objects.append(obj)
            except json.JSONDecodeError:
                continue
        
        return json_objects
    
    def get_structured_data(self):
        """获取解析的结构化数据"""
        return self.structured_data

# 使用结构化流式处理器
structured_handler = StructuredStreamingHandler()

# 创建提示，要求模型生成结构化数据
structured_prompt = """
请生成一个包含3个编程语言学习资源的列表，每个资源包含以下字段：
- name: 资源名称
- type: 资源类型(书籍/视频/网站)
- difficulty: 难度级别(初级/中级/高级)
- topics: 涵盖的主题列表

请以JSON格式返回，每个资源为一个JSON对象。
"""

chat_model = ChatOpenAI(
    streaming=True,
    callbacks=[structured_handler],
    temperature=0.3
)

message = HumanMessage(content=structured_prompt)
response = chat_model([message])

# 获取解析的结构化数据
structured_data = structured_handler.get_structured_data()
print(f"\n总共解析到 {len(structured_data)} 个结构化对象")

嵌入模型（embeddings model）

嵌入模型是 LangChain 中用于将文本转换为向量表示的关键组件，这些向量可以用于相似度计算、聚类、检索等任务。

通过OpenAIEmbeddings 使用 嵌入模型（Embedding Model）

1. 基本使用方法

from langchain.embeddings import OpenAIEmbeddings

# 初始化OpenAI嵌入模型
embeddings = OpenAIEmbeddings()

# 单个文本嵌入
text = "LangChain是一个用于构建语言模型应用的框架"
text_embedding = embeddings.embed_query(text)

print(f"嵌入向量维度: {len(text_embedding)}")
print(f"前5个维度: {text_embedding[:5]}")

2. 批量文本嵌入

# 多个文本批量嵌入
texts = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络进行学习",
    "自然语言处理处理人类语言",
    "计算机视觉处理图像和视频"
]

text_embeddings = embeddings.embed_documents(texts)

print(f"嵌入了 {len(text_embeddings)} 个文本")
print(f"每个嵌入向量维度: {len(text_embeddings[0])}")

3. 文本相似度计算

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# 计算文本相似度
query = "什么是机器学习？"
documents = [
    "机器学习是让计算机从数据中学习的技术",
    "深度学习是机器学习的一个子领域",
    "今天天气很好，适合出去散步",
    "人工智能包括机器学习和深度学习"
]

# 嵌入查询和文档
query_embedding = embeddings.embed_query(query)
doc_embeddings = embeddings.embed_documents(documents)

# 计算相似度
similarities = cosine_similarity([query_embedding], doc_embeddings)[0]

# 打印相似度结果
for i, (doc, sim) in enumerate(zip(documents, similarities)):
    print(f"文档 {i+1} (相似度: {sim:.4f}): {doc}")

# 找到最相似的文档
most_similar_idx = np.argmax(similarities)
print(f"\n最相似的文档: {documents[most_similar_idx]}")

4. 文档检索应用

from langchain.vectorstores import FAISS
from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader

# 加载文档
loader = TextLoader("example.txt")  # 假设有一个示例文档
documents = loader.load()

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

# 创建向量存储
vectorstore = FAISS.from_documents(texts, embeddings)

# 执行相似性搜索
query = "机器学习的主要应用领域有哪些？"
docs = vectorstore.similarity_search(query, k=3)

# 打印检索结果
print(f"查询: {query}\n")
print("检索到的相关文档:")
for i, doc in enumerate(docs):
    print(f"\n文档 {i+1}:")
    print(doc.page_content)

5. 自定义嵌入模型配置

# 自定义OpenAI嵌入模型参数
custom_embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002",  # 指定模型
    chunk_size=1000,                 # 批处理大小
    max_retries=3,                   # 最大重试次数
    request_timeout=60               # 请求超时时间
)

# 使用自定义配置的嵌入模型
text = "自定义配置的嵌入模型"
embedding = custom_embeddings.embed_query(text)
print(f"自定义嵌入向量维度: {len(embedding)}")

6. 嵌入模型的性能优化

import time
from functools import lru_cache

class CachedEmbeddings:
    """带缓存的嵌入模型包装器"""
    
    def __init__(self, base_embeddings, cache_size=1000):
        self.base_embeddings = base_embeddings
        self.cache_size = cache_size
        self._setup_cache()
    
    def _setup_cache(self):
        """设置LRU缓存"""
        self.embed_query_cached = lru_cache(maxsize=self.cache_size)(self._embed_query)
        self.embed_documents_cached = lru_cache(maxsize=self.cache_size)(self._embed_documents)
    
    def _embed_query(self, text: str):
        """实际的查询嵌入方法"""
        return self.base_embeddings.embed_query(text)
    
    def _embed_documents(self, texts: list):
        """实际的文档嵌入方法"""
        # 将列表转换为元组以便缓存
        texts_tuple = tuple(texts)
        return self.base_embeddings.embed_documents(list(texts_tuple))
    
    def embed_query(self, text: str):
        """带缓存的查询嵌入"""
        return self.embed_query_cached(text)
    
    def embed_documents(self, texts: list):
        """带缓存的文档嵌入"""
        return self.embed_documents_cached(texts)

# 使用缓存嵌入模型
base_embeddings = OpenAIEmbeddings()
cached_embeddings = CachedEmbeddings(base_embeddings, cache_size=500)

# 测试性能
texts = ["测试文本1", "测试文本2", "测试文本3"]

# 第一次调用 - 实际计算
start_time = time.time()
embeddings1 = cached_embeddings.embed_documents(texts)
first_call_time = time.time() - start_time

# 第二次调用相同文本 - 从缓存获取
start_time = time.time()
embeddings2 = cached_embeddings.embed_documents(texts)
second_call_time = time.time() - start_time

print(f"第一次调用时间: {first_call_time:.4f}秒")
print(f第二次调用时间: {second_call_time:.4f}秒")
print(f"性能提升: {first_call_time/second_call_time:.2f}倍")

7. 异步嵌入处理

import asyncio
from langchain.embeddings import OpenAIEmbeddings

async def async_embedding_example():
    """异步嵌入处理示例"""
    embeddings = OpenAIEmbeddings()
    
    # 异步嵌入单个文本
    text = "异步处理可以提高应用程序性能"
    embedding = await embeddings.aembed_query(text)
    print(f"异步嵌入向量维度: {len(embedding)}")
    
    # 异步嵌入多个文本
    texts = [
        "异步编程允许并发执行任务",
        "Python的asyncio库支持异步编程",
        "异步操作可以提高I/O密集型应用的性能"
    ]
    
    embeddings_list = await embeddings.aembed_documents(texts)
    print(f"异步嵌入了 {len(embeddings_list)} 个文本")

# 运行异步示例
asyncio.run(async_embedding_example())

8. 多语言嵌入处理

# 多语言文本嵌入示例
multilingual_texts = [
    "人工智能正在改变世界",           # 中文
    "Artificial intelligence is changing the world",  # 英文
    "La inteligencia artificial está cambiando el mundo",  # 西班牙语
    "L'intelligence artificielle change le monde"  # 法语
]

# 嵌入多语言文本
multilingual_embeddings = embeddings.embed_documents(multilingual_texts)

# 计算不同语言文本之间的相似度
similarities = cosine_similarity(multilingual_embeddings)

print("多语言文本相似度矩阵:")
for i, text in enumerate(multilingual_texts):
    print(f"\n文本 {i+1}: {text}")
    for j, sim in enumerate(similarities[i]):
        print(f"  与文本 {j+1} 的相似度: {sim:.4f}")

9. 嵌入模型与向量数据库集成

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import DirectoryLoader

# 加载文档目录
loader = DirectoryLoader('./documents/', glob="**/*.txt")
documents = loader.load()

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

# 创建嵌入模型
embeddings = OpenAIEmbeddings()

# 创建Chroma向量数据库
vectorstore = Chroma.from_documents(texts, embeddings, persist_directory="./chroma_db")
vectorstore.persist()

# 查询向量数据库
query = "机器学习在医疗领域的应用"
docs = vectorstore.similarity_search_with_score(query, k=5)

print(f"查询: {query}\n")
for i, (doc, score) in enumerate(docs):
    print(f"\n文档 {i+1} (相似度分数: {score:.4f}):")
    print(doc.page_content[:200] + "...")  # 只显示前200个字符

10. 嵌入质量评估

import numpy as np
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score

def evaluate_embedding_quality(embeddings, texts, labels=None):
    """评估嵌入质量"""
    # 如果没有提供标签，使用聚类生成标签
    if labels is None:
        n_clusters = min(5, len(texts) // 2)  # 确定聚类数量
        kmeans = KMeans(n_clusters=n_clusters, random_state=42)
        labels = kmeans.fit_predict(embeddings)
    
    # 计算轮廓系数 - 衡量聚类质量
    silhouette_avg = silhouette_score(embeddings, labels)
    
    # 计算类内平均距离
    intra_cluster_distances = []
    for cluster_id in np.unique(labels):
        cluster_embeddings = embeddings[labels == cluster_id]
        if len(cluster_embeddings) > 1:
            distances = cosine_similarity(cluster_embeddings)
            # 排除对角线上的1（自己与自己的相似度）
            np.fill_diagonal(distances, 0)
            intra_cluster_distances.append(np.mean(distances))
    
    avg_intra_similarity = np.mean(intra_cluster_distances) if intra_cluster_distances else 0
    
    return {
        "silhouette_score": silhouette_avg,
        "avg_intra_cluster_similarity": avg_intra_similarity,
        "n_clusters": len(np.unique(labels))
    }

# 评估嵌入质量
texts = [
    "机器学习是人工智能的子领域",
    "深度学习使用神经网络",
    "自然语言处理处理文本数据",
    "计算机视觉处理图像",
    "强化学习通过奖励学习",
    "监督学习使用标注数据",
    "无监督学习发现数据模式",
    "Python是一种编程语言",
    "Java也是编程语言",
    "JavaScript用于Web开发"
]

# 生成嵌入
text_embeddings = embeddings.embed_documents(texts)

# 评估质量
quality_metrics = evaluate_embedding_quality(text_embeddings, texts)
print("嵌入质量评估结果:")
for metric, value in quality_metrics.items():
    print(f"{metric}: {value:.4f}")

LLMS 大语言模型

嵌入模型和大模型的区别

嵌入模型和大语言模型虽然都是基于深度学习的NLP模型，但它们在设计目的、功能和应用场景上有显著区别。

1. 基本概念区别

嵌入模型 (Embedding Models):

• 专门用于将文本转换为数值向量表示
• 输出是固定维度的向量，通常用于相似度计算、聚类、检索等任务
• 不直接生成可读的文本
• 主要关注文本的语义表示

大语言模型 (Large Language Models, LLMs):

• 专门用于理解和生成人类语言
• 输入是文本提示，输出是生成的文本
• 能够进行对话、回答问题、创作内容等
• 主要关注语言的生成和理解

2. 架构和训练差异

# 嵌入模型示例
from langchain.embeddings import OpenAIEmbeddings

# 嵌入模型专注于将文本转换为向量
embeddings_model = OpenAIEmbeddings(model="text-embedding-ada-002")

text = "人工智能是未来的发展方向"
embedding_vector = embeddings_model.embed_query(text)

print(f"嵌入向量维度: {len(embedding_vector)}")
print(f"嵌入向量前5个值: {embedding_vector[:5]}")

# 大语言模型示例
from langchain.llms import OpenAI

# 大语言模型专注于文本生成
llm = OpenAI(model="text-davinci-003", temperature=0.7)

prompt = "请解释什么是人工智能："
generated_text = llm(prompt)

print(f"\n生成的文本: {generated_text}")

3. 功能对比

特性	嵌入模型	大语言模型
主要功能	文本向量化	文本理解与生成
输出形式	数值向量	可读文本
应用场景	相似度计算、检索、聚类	对话、问答、创作
计算复杂度	相对较低	相对较高
上下文处理	通常处理单个文本或短序列	可处理长上下文
训练目标	语义表示学习	语言建模

4. 使用场景对比

# 嵌入模型典型使用场景：语义搜索
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter

# 文档库
documents = [
    "机器学习是人工智能的一个分支，它使计算机能够从数据中学习。",
    "深度学习是机器学习的一个子集，使用神经网络模拟人脑。",
    "自然语言处理是AI的一个领域，专注于计算机与人类语言的交互。",
    "计算机视觉使机器能够理解和解释视觉信息。"
]

# 创建嵌入
embeddings = OpenAIEmbeddings()
doc_embeddings = embeddings.embed_documents(documents)

# 创建向量存储用于检索
vectorstore = FAISS.from_texts(documents, embeddings)

# 查询相似文档
query = "什么是深度学习？"
results = vectorstore.similarity_search(query, k=2)

print("嵌入模型应用 - 语义搜索:")
for doc in results:
    print(f"- {doc.page_content}")

# 大语言模型典型使用场景：问答生成
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 创建问答链
llm = OpenAI(temperature=0.7)
prompt = PromptTemplate(
    input_variables=["question"],
    template="请详细回答以下问题：{question}"
)
chain = LLMChain(llm=llm, prompt=prompt)

# 生成答案
question = "什么是深度学习？"
answer = chain.run(question)

print(f"\n大语言模型应用 - 问答生成:")
print(f"问题: {question}")
print(f"答案: {answer}")

5. 结合使用的最佳实践

from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import OpenAI
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.document_loaders import TextLoader

# 最佳实践：结合使用嵌入模型和大语言模型

# 1. 使用嵌入模型创建向量数据库
documents = [
    "机器学习是一种使计算机系统能够从经验中学习和改进的技术，而无需明确编程。",
    "深度学习是机器学习的一个子集，它使用多层神经网络来模拟人脑的工作方式。",
    "自然语言处理(NLP)是人工智能的一个分支，专注于计算机与人类语言之间的交互。",
    "强化学习是一种机器学习方法，智能体通过与环境交互来学习如何最大化累积奖励。"
]

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(documents, embeddings)

# 2. 使用大语言模型进行问答
llm = OpenAI(temperature=0.7)

# 3. 创建检索式问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 4. 执行问答
query = "什么是强化学习？"
result = qa_chain.run(query)

print("结合使用示例:")
print(f"查询: {query}")
print(f"回答: {result}")

# 工作流程说明
print("\n工作流程:")
print("1. 嵌入模型将查询转换为向量")
print("2. 在向量数据库中检索相关文档")
print("3. 将检索到的文档和查询一起提供给大语言模型")
print("4. 大语言模型基于上下文生成答案")

通过以上对比和分析，我们可以看到嵌入模型和大语言模型在AI应用中各有其独特价值：

• 嵌入模型擅长将文本转换为语义向量，是检索、聚类、相似度计算等任务的基础
• 大语言模型擅长理解和生成人类语言，是对话、问答、创作等任务的核心
• 最佳实践是结合两者的优势，使用嵌入模型进行语义检索，使用大语言模型进行内容生成

这种组合使用方式正是LangChain框架设计的核心理念，通过灵活组合不同的模型组件，构建强大的AI应用。