====== 第二章:核心组件详解 - Models ======
===== 2.1 模型类型概述 =====
在LangChain中,语言模型是最核心的组件。LangChain支持三大类模型:
==== 2.1.1 LLMs(基础语言模型) ====
**定义**:接收字符串输入,返回字符串输出的模型。
**特点**:
* 传统的文本补全接口
* 通常通过调用 `predict()` 或 `generate()` 方法
* 适合简单的文本生成任务
**代表模型**:
* OpenAI: `text-davinci-003`, `gpt-3.5-turbo-instruct`
* 开源: `LLaMA`, `Mistral`, `Falcon`
**使用场景**:
* 文本补全
* 简单的文本转换
* 需要直接控制输入格式的场景
from langchain.llms import OpenAI
# 创建LLM实例
llm = OpenAI(
model_name="gpt-3.5-turbo-instruct",
temperature=0.7,
max_tokens=256
)
# 生成文本
text = "人工智能正在改变"
result = llm.predict(text)
print(result)
==== 2.1.2 Chat Models(对话模型) ====
**定义**:专为对话场景优化的模型,接收消息列表,返回消息。
**特点**:
* 基于消息角色的架构(System、Human、AI)
* 更好的对话上下文理解
* 现代主流模型的标准接口
**代表模型**:
* OpenAI: `gpt-3.5-turbo`, `gpt-4`, `gpt-4-turbo`
* Anthropic: `claude-3-opus`, `claude-3-sonnet`, `claude-3-haiku`
* 开源: `LLaMA-2-Chat`, `Vicuna`
**使用场景**:
* 聊天机器人
* 多轮对话系统
* 需要角色设定的场景
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
chat = ChatOpenAI(model="gpt-4")
messages = [
SystemMessage(content="你是一个Python专家"),
HumanMessage(content="解释什么是装饰器")
]
response = chat.predict_messages(messages)
print(response.content)
==== 2.1.3 Embedding Models(嵌入模型) ====
**定义**:将文本转换为高维向量(嵌入)的模型。
**特点**:
* 输出是数值向量,不是文本
* 捕捉语义信息
* 用于相似度计算和检索
**代表模型**:
* OpenAI: `text-embedding-3-small`, `text-embedding-3-large`, `text-embedding-ada-002`
* 开源: `sentence-transformers`系列
**使用场景**:
* 语义搜索
* 文本聚类
* RAG系统中的文档检索
from langchain.embeddings import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 获取文本的向量表示
text = "这是一个测试句子"
vector = embeddings.embed_query(text)
print(f"向量维度: {len(vector)}")
print(f"向量前5个值: {vector[:5]}")
# 批量嵌入
texts = ["第一句", "第二句", "第三句"]
vectors = embeddings.embed_documents(texts)
print(f"批量嵌入数量: {len(vectors)}")
==== 2.1.4 三种模型的对比 ====
| 特性 | LLMs | Chat Models | Embeddings |
| 输入 | 字符串 | 消息列表 | 字符串 |
| 输出 | 字符串 | 消息对象 | 向量 |
| 主要用途 | 文本生成 | 对话 | 语义表示 |
| 状态管理 | 无 | 无 | 无 |
| 成本 | 按token计费 | 按token计费 | 通常较低 |
----
===== 2.2 OpenAI 模型集成 =====
==== 2.2.1 ChatOpenAI 详解 ====
`ChatOpenAI` 是与OpenAI聊天模型交互的主要接口。
=== 基础配置参数 ===
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
# 模型选择
model="gpt-4", # 模型名称
# 生成参数
temperature=0.7, # 随机性 (0-2)
max_tokens=None, # 最大输出token数
top_p=1.0, # 核采样概率
frequency_penalty=0.0, # 频率惩罚 (-2 to 2)
presence_penalty=0.0, # 存在惩罚 (-2 to 2)
# API配置
api_key="your-api-key", # API密钥
base_url=None, # 自定义API端点
timeout=None, # 请求超时时间
max_retries=2, # 最大重试次数
# 其他
streaming=False, # 是否流式输出
n=1, # 生成结果数量
stop=None, # 停止序列
)
=== 参数详解 ===
**1. temperature(温度)**
控制输出的随机性:
* `0.0`: 最确定性,总是选择概率最高的token
* `0.7`: 平衡值,适合大多数场景
* `1.0+`: 更具创造性,输出更多样
def demonstrate_temperature():
"""演示temperature参数的影响"""
prompt = "给一家咖啡店起一个有创意的名字"
for temp in [0.0, 0.5, 1.0, 1.5]:
llm = ChatOpenAI(temperature=temp)
result = llm.predict(prompt)
print(f"\nTemperature = {temp}:")
print(result)
demonstrate_temperature()
**2. max_tokens(最大token数)**
限制模型输出的长度:
* 一个token约等于4个英文字符或0.75个单词
* 中文通常1-2个字符为一个token
* 设置为None时,模型自行决定长度
# 不同max_tokens的效果
prompt = "简要介绍Python编程语言"
for max_tok in [50, 100, 200]:
llm = ChatOpenAI(max_tokens=max_tok)
result = llm.predict(prompt)
print(f"\nmax_tokens={max_tok}:")
print(result[:100] + "..." if len(result) > 100 else result)
**3. frequency_penalty 和 presence_penalty**
用于减少重复:
* `frequency_penalty`: 基于token出现频率的惩罚
* `presence_penalty`: 只要出现过的token都会受到惩罚
* 范围都是 -2.0 到 2.0
# 减少重复内容的示例
prompt = "请写一个关于'未来城市'的100字描述,不要重复用词"
llm_low = ChatOpenAI(frequency_penalty=0.0)
llm_high = ChatOpenAI(frequency_penalty=0.8)
print("低惩罚:")
print(llm_low.predict(prompt))
print("\n高惩罚:")
print(llm_high.predict(prompt))
==== 2.2.2 调用方法详解 ====
ChatOpenAI 提供多种调用方式:
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage
chat = ChatOpenAI()
# 方法1: predict - 最简单的方式
result = chat.predict("你好")
print(type(result)) # str
# 方法2: predict_messages - 使用消息对象
messages = [
SystemMessage(content="你是一个助手"),
HumanMessage(content="你好")
]
result = chat.predict_messages(messages)
print(type(result)) # AIMessage
print(result.content)
# 方法3: generate - 批量生成,获取更多元数据
from langchain.schema import Generation
batch_messages = [
[HumanMessage(content="你好")],
[HumanMessage(content="再见")]
]
result = chat.generate(batch_messages)
print(f"总token消耗: {result.llm_output['token_usage']}")
print(f"第一个结果: {result.generations[0][0].text}")
# 方法4: async/await - 异步调用
import asyncio
async def async_chat():
result = await chat.apredict("异步测试")
print(result)
asyncio.run(async_chat())
# 方法5: streaming - 流式输出
chat_stream = ChatOpenAI(streaming=True)
for chunk in chat_stream.stream("写一首短诗"):
print(chunk.content, end="", flush=True)
==== 2.2.3 模型选择指南 ====
OpenAI提供了多个模型,如何选择?
| 模型 | 优点 | 缺点 | 适用场景 |
| gpt-3.5-turbo | 快、便宜 | 能力有限 | 简单任务、原型开发 |
| gpt-4 | 能力强 | 慢、贵 | 复杂推理、代码生成 |
| gpt-4-turbo | 能力最强、支持长文本 | 最贵 | 高级应用、长文档处理 |
| gpt-4o | 多模态、快 | 较新 | 需要图像理解的场景 |
def select_model_for_task(task_type: str) -> str:
"""根据任务类型推荐模型"""
models = {
"简单问答": "gpt-3.5-turbo",
"代码生成": "gpt-4",
"复杂推理": "gpt-4-turbo",
"创意写作": "gpt-4",
"文本分类": "gpt-3.5-turbo",
"多语言翻译": "gpt-4",
"数学计算": "gpt-4-turbo",
}
return models.get(task_type, "gpt-3.5-turbo")
# 使用示例
tasks = ["简单问答", "代码生成", "创意写作"]
for task in tasks:
model = select_model_for_task(task)
print(f"{task} -> {model}")
==== 2.2.4 OpenAI Embedding 模型 ====
from langchain_openai import OpenAIEmbeddings
# 选择不同的嵌入模型
embeddings_small = OpenAIEmbeddings(model="text-embedding-3-small") # 1536维
embeddings_large = OpenAIEmbeddings(model="text-embedding-3-large") # 3072维
# 对比不同模型
test_text = "自然语言处理是人工智能的重要分支"
vector_small = embeddings_small.embed_query(test_text)
vector_large = embeddings_large.embed_query(test_text)
print(f"text-embedding-3-small: {len(vector_small)} 维")
print(f"text-embedding-3-large: {len(vector_large)} 维")
# 计算两个句子的相似度
import numpy as np
def cosine_similarity(a, b):
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
text1 = "机器学习是AI的子领域"
text2 = "深度学习是机器学习的一种方法"
text3 = "今天天气很好"
v1 = embeddings_small.embed_query(text1)
v2 = embeddings_small.embed_query(text2)
v3 = embeddings_small.embed_query(text3)
print(f"\n'{text1}' 和 '{text2}' 的相似度: {cosine_similarity(v1, v2):.4f}")
print(f"'{text1}' 和 '{text3}' 的相似度: {cosine_similarity(v1, v3):.4f}")
----
===== 2.3 Anthropic Claude 模型集成 =====
Claude是Anthropic开发的AI助手,以安全性和有用性著称。
==== 2.3.1 ChatAnthropic 基础使用 ====
from langchain_anthropic import ChatAnthropic
from langchain.schema import HumanMessage, SystemMessage
# 创建Claude客户端
claude = ChatAnthropic(
model="claude-3-sonnet-20240229",
temperature=0.7,
max_tokens=1024,
anthropic_api_key="your-api-key" # 或使用环境变量 ANTHROPIC_API_KEY
)
# 基本对话
messages = [
SystemMessage(content="你是一个专业的技术文档写手"),
HumanMessage(content="写一段关于REST API的介绍")
]
response = claude.predict_messages(messages)
print(response.content)
==== 2.3.2 Claude 模型对比 ====
| 模型 | 描述 | 最佳用途 |
| claude-3-opus | 最强大 | 复杂推理、数学、编程 |
| claude-3-sonnet | 平衡 | 大多数任务,性价比高 |
| claude-3-haiku | 最快 | 简单任务、实时应用 |
def compare_claude_models():
"""对比不同Claude模型的表现"""
models = [
"claude-3-haiku-20240307",
"claude-3-sonnet-20240229",
"claude-3-opus-20240229"
]
prompt = "用一句话解释量子计算"
for model in models:
print(f"\n{'='*50}")
print(f"模型: {model}")
print('='*50)
llm = ChatAnthropic(model=model)
result = llm.predict(prompt)
print(result)
compare_claude_models()
==== 2.3.3 Claude 的特殊功能 ====
Claude支持一些特殊功能:
from langchain_anthropic import ChatAnthropic
claude = ChatAnthropic(model="claude-3-sonnet-20240229")
# 1. 长上下文窗口(200K tokens)
long_text = "请总结以下长文档..." + "..." * 100000
response = claude.predict(long_text[:150000]) # Claude支持超长输入
# 2. 结构化提示
structured_prompt = """
Human: 请分析以下产品评价,输出JSON格式:
评价:"这个手机电池续航很好,但屏幕有点暗"
请按以下格式输出:
{
"sentiment": "positive/negative/mixed",
"aspects": [
{"aspect": "...", "sentiment": "...", "comment": "..."}
]
}
Assistant:"""
result = claude.predict(structured_prompt)
print(result)
----
===== 2.4 本地模型集成 =====
==== 2.4.1 使用 HuggingFace 模型 ====
LangChain可以通过HuggingFace集成各种开源模型。
from langchain_community.llms import HuggingFacePipeline
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
import torch
# 方法1: 直接使用HuggingFace Pipeline
def load_local_model():
model_id = "microsoft/DialoGPT-medium" # 可以替换为其他模型
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(model_id)
pipe = pipeline(
"text-generation",
model=model,
tokenizer=tokenizer,
max_length=100,
temperature=0.7
)
llm = HuggingFacePipeline(pipeline=pipe)
return llm
# 使用本地模型
local_llm = load_local_model()
result = local_llm.predict("你好")
print(result)
==== 2.4.2 使用 llama.cpp / llama-cpp-python ====
对于消费级硬件,llama.cpp是很好的选择。
from langchain_community.llms import LlamaCpp
# 加载GGUF格式的模型
llm = LlamaCpp(
model_path="path/to/llama-2-7b-chat.Q4_K_M.gguf",
n_ctx=2048, # 上下文窗口大小
n_gpu_layers=1, # 使用GPU的层数
temperature=0.7,
max_tokens=512,
verbose=True
)
# 生成文本
result = llm.predict("解释什么是机器学习")
print(result)
==== 2.4.3 使用 Ollama ====
Ollama让在本地运行大模型变得非常简单。
from langchain_community.llms import Ollama
# 连接到本地Ollama服务
ollama = Ollama(
model="llama2", # 或 "mistral", "codellama" 等
base_url="http://localhost:11434"
)
# 使用
result = ollama.predict("你好,请介绍一下自己")
print(result)
# 也可以这样创建
llm = Ollama(model="llama2:13b")
==== 2.4.4 本地嵌入模型 ====
from langchain_community.embeddings import HuggingFaceEmbeddings
# 使用开源嵌入模型
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2",
model_kwargs={'device': 'cpu'}, # 或 'cuda'
encode_kwargs={'normalize_embeddings': True}
)
# 使用
text = "这是一个测试句子"
vector = embeddings.embed_query(text)
print(f"向量维度: {len(vector)}")
----
===== 2.5 其他模型提供商 =====
==== 2.5.1 Azure OpenAI ====
from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings
# Azure OpenAI配置
llm = AzureChatOpenAI(
azure_endpoint="https://your-resource.openai.azure.com/",
azure_deployment="your-deployment-name",
openai_api_version="2024-02-01",
openai_api_key="your-api-key"
)
embeddings = AzureOpenAIEmbeddings(
azure_endpoint="https://your-resource.openai.azure.com/",
azure_deployment="your-embedding-deployment",
openai_api_version="2024-02-01"
)
==== 2.5.2 Google Vertex AI / Gemini ====
from langchain_google_vertexai import ChatVertexAI
from langchain_google_genai import ChatGoogleGenerativeAI
# Vertex AI
gemini = ChatVertexAI(
model_name="gemini-pro",
project="your-project-id",
location="us-central1"
)
# 或使用Google Generative AI
gemini = ChatGoogleGenerativeAI(
model="gemini-pro",
google_api_key="your-api-key"
)
result = gemini.predict("你好")
print(result)
==== 2.5.3 国产大模型 ====
# 文心一言
from langchain_community.chat_models import QianfanChatEndpoint
wenxin = QianfanChatEndpoint(
qianfan_ak="your-access-key",
qianfan_sk="your-secret-key"
)
# 通义千问
from langchain_community.chat_models import Tongyi
tongyi = Tongyi(
dashscope_api_key="your-api-key"
)
# 讯飞星火
from langchain_community.chat_models import SparkLLM
spark = SparkLLM(
spark_app_id="your-app-id",
spark_api_key="your-api-key",
spark_api_secret="your-secret"
)
----
===== 2.6 模型进阶技巧 =====
==== 2.6.1 模型降级策略 ====
当主要模型不可用时自动切换到备用模型:
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.schema import BaseMessage
class FallbackLLM:
"""带降级策略的LLM包装器"""
def __init__(self, primary, fallback):
self.primary = primary
self.fallback = fallback
def predict(self, text: str) -> str:
try:
return self.primary.predict(text)
except Exception as e:
print(f"Primary model failed: {e}, falling back...")
return self.fallback.predict(text)
def predict_messages(self, messages: list[BaseMessage]) -> str:
try:
result = self.primary.predict_messages(messages)
return result.content
except Exception as e:
print(f"Primary model failed: {e}, falling back...")
result = self.fallback.predict_messages(messages)
return result.content
# 使用
primary = ChatOpenAI(model="gpt-4")
fallback = ChatOpenAI(model="gpt-3.5-turbo")
llm_with_fallback = FallbackLLM(primary, fallback)
result = llm_with_fallback.predict("你好")
print(result)
==== 2.6.2 模型路由器 ====
根据输入自动选择最合适的模型:
from langchain_openai import ChatOpenAI
class ModelRouter:
"""基于任务类型路由到不同模型"""
def __init__(self):
self.models = {
"simple": ChatOpenAI(model="gpt-3.5-turbo"),
"complex": ChatOpenAI(model="gpt-4"),
"code": ChatOpenAI(model="gpt-4", temperature=0.2),
}
self.routing_prompt = """分析以下查询,分类为:simple(简单问答)、complex(复杂推理)、code(代码相关)。
只输出类别词。
查询: {query}
类别:"""
self.classifier = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
def classify(self, query: str) -> str:
result = self.classifier.predict(self.routing_prompt.format(query=query))
return result.strip().lower()
def predict(self, query: str) -> str:
category = self.classify(query)
print(f"分类结果: {category}")
model = self.models.get(category, self.models["simple"])
return model.predict(query)
# 使用
router = ModelRouter()
queries = [
"你好", # simple
"解释量子纠缠的物理原理", # complex
"写一个Python函数计算斐波那契数列" # code
]
for q in queries:
print(f"\n查询: {q}")
result = router.predict(q)
print(f"回复: {result[:100]}...")
==== 2.6.3 批量请求优化 ====
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
import asyncio
async def batch_predict(queries: list[str], batch_size: int = 5):
"""批量异步预测"""
llm = ChatOpenAI()
results = []
for i in range(0, len(queries), batch_size):
batch = queries[i:i+batch_size]
# 准备消息
messages_list = [[HumanMessage(content=q)] for q in batch]
# 批量生成
batch_results = await llm.agenerate(messages_list)
for gen in batch_results.generations:
results.append(gen[0].text)
return results
# 使用示例
async def main():
queries = [f"问题{i}: 解释什么是Python" for i in range(10)]
results = await batch_predict(queries)
for q, r in zip(queries, results):
print(f"{q} -> {r[:50]}...")
asyncio.run(main())
==== 2.6.4 缓存机制 ====
避免重复调用API,节省成本和延迟:
from langchain.globals import set_llm_cache
from langchain.cache import InMemoryCache, SQLiteCache
from langchain_openai import ChatOpenAI
# 内存缓存
set_llm_cache(InMemoryCache())
# 或持久化缓存
# set_llm_cache(SQLiteCache(database_path=".langchain.db"))
llm = ChatOpenAI()
# 第一次调用,会访问API
result1 = llm.predict("你好")
print("第一次调用完成")
# 第二次调用同样的输入,直接从缓存返回
result2 = llm.predict("你好")
print("第二次调用完成(使用缓存)")
# 清除缓存
# from langchain.globals import get_llm_cache
# get_llm_cache().clear()
----
===== 2.7 最佳实践 =====
==== 2.7.1 API密钥管理 ====
# config.py - 集中管理配置
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
# 模型默认配置
DEFAULT_MODEL = "gpt-3.5-turbo"
DEFAULT_TEMPERATURE = 0.7
@classmethod
def validate(cls):
"""验证必要配置"""
if not cls.OPENAI_API_KEY:
raise ValueError("OPENAI_API_KEY not set")
# 使用
from config import Config
from langchain_openai import ChatOpenAI
Config.validate()
llm = ChatOpenAI(api_key=Config.OPENAI_API_KEY)
==== 2.7.2 错误处理 ====
from langchain_openai import ChatOpenAI
from openai import RateLimitError, AuthenticationError
import time
def robust_predict(llm, prompt, max_retries=3):
"""健壮的预测函数,带重试机制"""
for attempt in range(max_retries):
try:
return llm.predict(prompt)
except RateLimitError:
wait_time = 2 ** attempt # 指数退避
print(f"Rate limit hit, waiting {wait_time}s...")
time.sleep(wait_time)
except AuthenticationError as e:
print(f"Authentication failed: {e}")
raise
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
raise
return None
# 使用
llm = ChatOpenAI()
result = robust_predict(llm, "你好")
==== 2.7.3 成本监控 ====
from langchain.callbacks import get_openai_callback
from langchain_openai import ChatOpenAI
def track_cost(llm, prompts: list[str]):
"""追踪API调用成本"""
with get_openai_callback() as cb:
results = []
for prompt in prompts:
result = llm.predict(prompt)
results.append(result)
print(f"总调用次数: {cb.successful_requests}")
print(f"总token数: {cb.total_tokens}")
print(f"提示token: {cb.prompt_tokens}")
print(f"完成token: {cb.completion_tokens}")
print(f"预估成本: ${cb.total_cost:.4f}")
return results
# 使用
llm = ChatOpenAI(model="gpt-4")
prompts = ["问题1", "问题2", "问题3"]
results = track_cost(llm, prompts)
----
===== 2.8 本章小结 =====
==== 核心概念回顾 ====
- **三种模型类型**
* LLMs: 文本补全模型
* Chat Models: 对话模型(最常用)
* Embeddings: 文本向量模型
- **主要集成**
* OpenAI: gpt-3.5-turbo, gpt-4系列
* Anthropic: Claude 3系列
* 本地: HuggingFace, Ollama, llama.cpp
* 云服务: Azure, Vertex AI
- **关键参数**
* temperature: 控制创造性
* max_tokens: 控制输出长度
* frequency/presence_penalty: 减少重复
- **进阶技巧**
* 模型降级策略
* 智能路由
* 批量请求
* 结果缓存
==== 选择决策树 ====
选择什么模型?
├── 需要最高质量?
│ ├── 是 → GPT-4 / Claude-3-Opus
│ └── 否 → 继续
├── 成本敏感?
│ ├── 是 → GPT-3.5-turbo / Claude-3-Haiku
│ └── 否 → 继续
├── 数据隐私要求高?
│ ├── 是 → 本地模型(Ollama/llama.cpp)
│ └── 否 → GPT-4 / Claude-3-Sonnet
└── 需要超长上下文?
├── 是 → Claude-3(200K) / GPT-4-turbo(128K)
└── 否 → 其他模型
==== 作业 ====
- 实现一个模型对比工具,对同一个问题比较不同模型的回答
- 为你的应用设计一个智能模型路由系统
- 实现一个带成本预算限制的LLM调用器
- 测试本地模型的部署和调用